11.9 spark SQL读取MySQL数据

绝地灬酷狼 2022-05-16 03:13 186阅读 0赞

注意别忘记了驱动包：mysql-connector-java-5.1.6.jar

public static void main(String\[\] args) \{

SparkConf conf = new SparkConf().setAppName("JDBCDataSource").setMaster("local");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

// 方法1、分别将mysql中两张表的数据加载为DataFrame

/\*

\* Map<String, String> options = new HashMap<String, String>();

\* options.put("url", "jdbc:mysql://hadoop1:3306/testdb");

\* options.put("driver", "com.mysql.jdbc.Driver");

\* options.put("user","spark");

\* options.put("password", "spark2016");

\* options.put("dbtable", "student\_info");

\* DataFrame studentInfosDF = sqlContext.read().format("jdbc").options(options).load();

\*

\* options.put("dbtable", "student\_score");

\* DataFrame studentScoresDF = sqlContext.read().format("jdbc") .options(options).load();

\*/

// 方法2、分别将mysql中两张表的数据加载为DataFrame

DataFrameReader reader = sqlContext.read().format("jdbc");

reader.option("url", "jdbc:mysql://hadoop1:3306/testdb");

reader.option("dbtable", "student\_info");

reader.option("driver", "com.mysql.jdbc.Driver");

reader.option("user", "spark");

reader.option("password", "spark2016");

DataFrame studentInfosDF = reader.load();

reader.option("dbtable", "student\_score");

DataFrame studentScoresDF = reader.load();

// 将两个DataFrame转换为JavaPairRDD，执行join操作

studentInfosDF.registerTempTable("studentInfos");

studentScoresDF.registerTempTable("studentScores");

String sql = "SELECT studentInfos.name,age,score "

+ " FROM studentInfos JOIN studentScores"

+ " ON (studentScores.name = studentInfos.name)"

+ " WHERE studentScores.score > 80";

DataFrame sql2 = sqlContext.sql(sql);

sql2.show();

//分析好的数据插入回数据库中，注意这边会创建很多个数据库链接，用foreachPartition优化

sql2.javaRDD().foreach(new VoidFunction<Row>() \{

private static final long serialVersionUID = 1L;

@Override

public void call(Row row) throws Exception \{

String sql = "insert into good\_student\_info values(" + "'" + String.valueOf(row.getString(0)) + "',"

+ Integer.valueOf(String.valueOf(row.get(1))) + ","

+ Integer.valueOf(String.valueOf(row.get(2))) + ")";

Class.forName("com.mysql.jdbc.Driver");

Connection conn = null;

Statement stmt = null;

try \{

conn = DriverManager.getConnection("jdbc:mysql://hadoop1:3306/testdb", "spark", "spark2016");

stmt = conn.createStatement();

stmt.executeUpdate(sql);

\} catch (Exception e) \{

e.printStackTrace();

\} finally \{

if (stmt != null) \{

stmt.close();

\}

if (conn != null) \{

conn.close();

\}

\}

\}

\});

/\*\*

\* 将SparkContext 关闭，释放资源

\*/

sc.close();

\}

发表评论取消回复

表情：

评论列表（有 0 条评论，186人围观）

还没有评论，来说两句吧...

相关阅读

相关 spark sql 读取 parquet 文件

SQL 你可以使用spark-sql -S启动 sql的 spark。 CREATE TEMPORARY VIEW parquetTable US

逃离我推掉我的手/ 2022年12月19日 06:12/ 0 赞/ 150 阅读

相关 Spark SQL读取hive数据时报找不到mysql驱动

Exception： Caused by: org.datanucleus.exceptions.NucleusException: Attempt to invoke th

谁践踏了优雅/ 2022年05月21日 03:43/ 0 赞/ 178 阅读

相关 SPARK Spark SQL读取hive数据时报找不到mysql驱动

错误提示：Attempt to invoke the "DBCP" plugin to create a ConnectionPool gave an error `将

心已赠人/ 2022年05月19日 12:11/ 0 赞/ 186 阅读

相关 11.9 spark SQL读取MySQL数据

注意别忘记了驱动包：mysql-connector-java-5.1.6.jar public static void main(String\[\] args) \{

绝地灬酷狼/ 2022年05月16日 03:13/ 0 赞/ 187 阅读

相关 11.8 spark SQL读取hive数据

1，启动hdfs集群 2，启动hive create database testdb 创建testdb库关闭hive保留关键字检查功能： set hive.suppo

Dear 丶/ 2022年05月16日 00:20/ 0 赞/ 184 阅读

相关 Spark连接MySQL数据库并读取数据

（作者：陈玓玏）打开pyspark，带驱动的那种用命令行启动pyspark时需要加上jdbc的驱动路径： pyspark --driver-class-p

短命女/ 2022年04月14日 05:06/ 0 赞/ 385 阅读

相关 Spark读取Hbase数据

\[size=medium\] 直接上代码：可以上传jar包，给远程的spark集群，这样能在idea中windows环境上运行代码： \[/size\]

以你之姓@/ 2022年04月12日 12:27/ 0 赞/ 296 阅读

相关 Intellj IDEA +SBT + Scala + Spark Sql读取HDFS数据

\[size=large\]前提Spark集群已经搭建完毕,如果不知道怎么搭建，请参考这个链接： \[url\]http://qindongliang.iteye.com/

ゝ一纸荒年。/ 2022年04月12日 10:22/ 0 赞/ 233 阅读

相关 Spark读取MySQL数据

import java.util.\{Properties, UUID\} import org.apache.spark.sql.\{SaveMode, SparkSessi

妖狐艹你老母/ 2021年10月13日 02:53/ 0 赞/ 373 阅读

相关 Spark并发读取MySQL

> Spark读取MySQL数据量过大，一直停在Added broadcast\_0\_piece0 in memory on cdh-master问题。 > > 19/09

朱雀/ 2021年09月23日 07:58/ 0 赞/ 231 阅读