文章目录
- 零、本讲学习目标
- 一、Spark SQL读取关系数据库
- 二、Spark SQL JDBC连接属性
- 三、创建数据库与表
- (一)创建数据库
- (二)创建学生表
- (二)创建成绩表
- 四、读取和写入数据库表
- (一)利用`dbtable`属性读取数据表
- (二)利用`dbtable`属性读取数据表查询
- (三)将数据帧内容写入数据表
- (四)利用`query`属性读取数据表查询
零、本讲学习目标
- 了解Spark SQL JDBC连接属性
- 掌握利用Spark SQL读取关系数据库
一、Spark SQL读取关系数据库
- Spark SQL还可以使用JDBC API从其他关系型数据库读取数据,返回的结果仍然是一个DataFrame,可以很容易地在Spark SQL中处理,或者与其他数据源进行连接查询。
二、Spark SQL JDBC连接属性
- 在使用JDBC连接数据库时可以指定相应的连接属性
属性 | 介绍 |
---|---|
url | 连接的JDBC URL |
driver | JDBC驱动的类名 |
user | 数据库用户名 |
password | 数据库密码 |
dbtable | 数据库表名或能代表一张数据库表的子查询。在读取数据时,若只使用数据库表名,则将查询整张表的数据;若希望查询部分数据或多表关联查询,则可以使用SQL查询的FROM子句中有效的任何内容,例如放入括号中的子查询。该属性的值会被当作一张表进行查询,查询格式:select * from <dbtable属性值> where 1 = 1。注意,不允许同时指定dbtable和query属性。 |
query | 指定查询的SQL语句。注意:不允许同时指定dbtable和query属性,也不允许同时指定query和partitionColumn属性。当需要指定partitionColumn属性时,可以使用dbtable属性指定子查询,并使用子查询的别名对分区列进行限定。 |
partitionColumn lowerBound upperBound | 这几个属性,若有一个被指定,则必须全部指定,且必须指定numPartitions属性。它们描述了如何在从多个Worker中并行读取数据时对表进行分区。partitionColumn必须是表中的数字、日期或时间戳列。注意,lowerBound 和upperBound只是用来决定分区跨度的,而不是用来过滤表中的行。因此,表中的所有行都将被分区并返回。 |
numPartitions | 对表并行读写数据时的最大分区数,这也决定了并发JDBC连接的最大数量。如果要写入数据的分区数量超过了此限制的值,那么在写入之前可以调用coalesce(numpartition)将分区数量减少到此限制的值。 |
三、创建数据库与表
(一)创建数据库
- 创建数据库
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);
(二)创建成绩表
- 创建表
student
,执行命令:CREATE TABLE score (id INT, name VARCHAR(10), score REAL);
- 给
score
表插入几条记录
四、读取和写入数据库表
- 启动Spark Shell
(一)利用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()
- 执行上述命令
- 执行命令:
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()
- 执行上述命令(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()
- 执行上述命令
- 在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()
-
执行上述命令,报错
-
Spark2.1.1的Spark SQL的JDBC属性里没有
query
。
-
从Spark 2.4.0开始的Spark SQL的JDBC属性里有
query
-
将Spark集群的Spark版本升级到Spark2.4.4
-
启动Spark Shell,执行命令:
spark-shell --master spark://master:7077
-
读取
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()
- 执行上述命令
- 执行命令:
resultDF.show()