零、本讲学习目标

1. 了解Spark SQL JDBC连接属性
2. 掌握利用Spark SQL读取关系数据库

一、Spark SQL读取关系数据库

• Spark SQL还可以使用JDBC API从其他关系型数据库读取数据，返回的结果仍然是一个DataFrame，可以很容易地在Spark SQL中处理，或者与其他数据源进行连接查询。

二、Spark SQL JDBC连接属性

• 在使用JDBC连接数据库时可以指定相应的连接属性

三、创建数据库与表

（一）创建数据库

• 创建数据库spark_db
  
  

（二）创建学生表

• 创建表student，执行命令：CREATE TABLE student (id INT, name VARCHAR(10), gender VARCHAR(2), age INT);
  
• 给student表插入几条记录
  

INSERT INTO student VALUES (1, '李文君', '女', 18);
INSERT INTO student VALUES (2, '唐玉龙', '男', 19);
INSERT INTO student VALUES (3, '陈燕文', '女', 20);
INSERT INTO student VALUES (4, '洪小刚', '男', 18);
INSERT INTO student VALUES (5, '郑小翠', '女', 19);

  

 

  
1
  
2
  
3
  
4
  
5
 

（二）创建成绩表

• 创建表student，执行命令：CREATE TABLE score (id INT, name VARCHAR(10), score REAL);
  
• 给score表插入几条记录
  

四、读取和写入数据库表

（一）利用dbtable属性读取数据表

• 读取student表

val studentDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://master:3306/spark_db?useSSL=false")     
  .option("driver","com.mysql.jdbc.Driver")  
  .option("dbtable", "student")  
  .option("user", "root")  
  .option("password", "903213")  
  .load()

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
 

• 执行上述命令
  
• 执行命令：studentDF.show()
  

（二）利用dbtable属性读取数据表查询

• 读取student与score关联查询结果

val resultDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://master:3306/spark_db?useSSL=false")     
  .option("driver","com.mysql.jdbc.Driver")  
  .option("dbtable", "(select st.id, st.name, gender, age, score from student st inner join score sc on st.id = sc.id) t")  
  .option("user", "root")  
  .option("password", "903213")  
  .load()

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
 

• 执行上述命令（dbtable属性的值是一个子查询，相当于SQL查询中的FROM关键字后的一部分）
  
• 查看结果数据帧内容，执行命令：resultDF.show()
  
• 将数据帧内容以json格式写入HDFS的/out目录
  
• 在slave1虚拟机上查看生成的json文件
  

（三）将数据帧内容写入数据表

• 将数据帧内容以jdbc格式写入数据库spark_db的test表

resultDF.write.mode(SaveMode.Overwrite).format("jdbc")
  .option("url", "jdbc:mysql://master:3306/spark_db?useSSL=false")     
  .option("dbtable", "test")
  .option("user", "root")
  .option("password", "903213")
  .save()

  

 

  
1
  
2
  
3
  
4
  
5
  
6
 

• 执行上述命令
  
• 在Navicat里查看生成的test表
  

（四）利用query属性读取数据表查询

• 读取student与score关联查询结果

val resultDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://master:3306/spark_db?useSSL=false")     
  .option("driver","com.mysql.jdbc.Driver")  
  .option("query", "select st.name, st.gender, sc.score from student st inner join score sc on st.id = sc.id")  
  .option("user", "root")  
  .option("password", "903213")  
  .load()

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
 

• 执行上述命令，报错
  
• 原因在于Spark2.1.1的Spark SQL的JDBC属性里没有query。
  
• 经查Spark 3.2.1的Spark SQL的JDBC属性里就有query。
  

文章来源: howard2005.blog.csdn.net，作者：howard2005，版权归原作者所有，如需转载，请联系作者。

原文链接：howard2005.blog.csdn.net/article/details/124742425