Spark SQL 连接MySQL
1、基本概念和用法（摘自Spark官方文档中文版）
其实该部分翻译自Spark官方文档，所以对于翻译有疑问的可直接看官方文档
Spark SQL 还有一个能够使用 JDBC 从其他数据库读取数据的数据源。当使用 JDBC 访问其它数据库时，应该 JdbcRDD。这是因为结果是以数据框（DataFrame）返回的，且这样 Spark SQL操作轻松或便于连接其它数据源。因为这种 JDBC 数据源不需要用户提供 ClassTag，所以它也更适合使用 Java 或 Python 操作。（注意，这与允许其它应用使用 Spark SQL 执行查询操作的 Spark SQL JDBC 服务器是不同的）。
使用 JDBC 访问特定数据库时，需要在 spark classpath 上添加对应的 JDBC 驱动配置。例如，为了从 Spark Shell 连接 postgres，你需要运行如下命令 :
bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar
通过调用数据源API，远程数据库的表可以被加载为DataFrame 或Spark SQL临时表。支持的参数有 :
2、Scala代码实现连接MySQL
2.1 添加MySQL 依赖
在sbt 配置文件里添加：
"mysql" % "mysql-connector-java" % "6.0.6"
然后执行：
sbt eclipse
2.2 建表并初始化数据
DROP TABLE IF EXISTS `USER_T`;
CREATE TABLE `USER_T` (
`ID` INT(11) NOT NULL,
`USER_NAME` VARCHAR(40) NOT NULL,
PRIMARY KEY (`ID`)
) ENGINE=INNODB DEFAULT CHARSET=UTF8;
INSERT INTO `USER_T`(`ID`,`USER_NAME`) VALUES (1,'测试1');
INSERT INTO `USER_T`(`ID`,`USER_NAME`) VALUES (2,'测试2');
2.3 代码
2.3.1 查询
package com.dkl.leanring.spark.sql
import org.apache.spark.sql.SparkSession
/**
* spark查询mysql测试
*/
object MysqlQueryDemo {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("MysqlQueryDemo").master("local").getOrCreate()
val jdbcDF = spark.read
.format("jdbc")
.option("url", "jdbc:mysql://192.168.44.128:3306/hive?useUnicode=true&characterEncoding=utf-8")
.option("dbtable", "USER_T")
.option("user", "root")
.option("password", "Root-123456")
.load()
jdbcDF.show()
}
}
2.3.2 插入数据
新建USER_T.csv,造几条数据如图：
（需将csv的编码格式转为utf-8,否则spark读取中文乱码，转码方法见：https://jingyan.baidu.com/article/fea4511a092e53f7bb912528.html）
package com.dkl.leanring.spark.sql
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode
import java.util.Properties
/**
* 从USER_T.csv读取数据并插入的mysql表中
*/
object MysqlInsertDemo {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("MysqlInsertDemo").master("local").getOrCreate()
val df = spark.read.option("header", "true").csv("src/main/resources/scala/USER_T.csv")
df.show()
val url = "jdbc:mysql://192.168.44.128:3306/hive?useUnicode=true&characterEncoding=utf-8"
val prop = new Properties()
prop.put("user", "root")
prop.put("password", "Root-123456")
df.write.mode(SaveMode.Append).jdbc(url, "USER_T", prop)
}
}
再查询一次，就会发现表里多了几条数据
3、注意（更新）
上面的代码在本地eclipse运行是没有问题的,如果放在服务器上用spark-submit提交的话，可能会报异常
java.sql.SQLException:No suitable driver
解决方法是在代码里添加
mysql:
.option("driver", "com.mysql.jdbc.Driver")
oracle:
.option("driver", "oracle.jdbc.driver.OracleDriver")
具体可参考我的另一篇博客：spark-submit报错:Exception in thread "main" java.sql.SQLException:No suitable driver
4、其他读取MySQL的方法（更新于2018.08.22）
发现还有Spark还有其他读取mysql的方法，其实只是上面讲的快捷方式（2.3.1），只附上用法，具体看以看api或参考其他博客，如spark：scala读取mysql的4种方法
(我就是在这篇博客看到的，该博客是Spark1.x的写法，若用Spark2.x将sqlContext改为本文的spark即可)
spark.read.jdbc(url, table, columnName, lowerBound, upperBound, numPartitions, connectionProperties)
spark.read.jdbc(url, table, predicates, connectionProperties)
spark.read.jdbc(url, table, properties)
只要在2.3.1的代码里用.option(key,value)即可
5、关于读取MySQL的分区设置（更新于2018.08.22）
按照2.3.1的代码读取的DataFrame的分区数为1，若想改变分区数，一种方法是重分区
df.repartition(numPartitions)
另一种是在读的时候设置分区数，在部分可以看到，通过numPartitions可以设置分区数，但是注意partitionColumn, lowerBound, upperBound, numPartitions需要同时设置
如：
val jdbcDF = spark.read
.format("jdbc")
.option("url", "jdbc:mysql://192.168.44.128:3306/hive?useUnicode=true&characterEncoding=utf-8")
.option("dbtable", "USER_T")
.option("user", "root")
.option("password", "Root-123456")
.option("numPartitions", "160")
.option("partitionColumn", "ID")
.option("lowerBound", "1")
.option("upperBound", "1000")
.load()
6、关于partitionColumn、lowerBound、upperBound（更新于2018.09.07）
一般只有当关系型数据库表数据量很大时，才会通过设置上面的参数设置分区，否则读进来再repartition即可，设置分区数的目的是可以使Spark读取数据库更快一点，经验证，合理的设置分区，Spark是可以并行的读取关系型数据库的。
partitionColumn是用于进行分区的字段，一般好是表的索引字段，且要求为字段类型为数字，并好是连续的，这样分区才不会数据倾斜，也就是每个并行的任务读取的数量级差不多，才会更快的读取大表。
lowerBound upperBound官网说是用来决定分区跨度的，对于这个说法我一开始理解的不太准确，它的值为partitionColumn在表里具体的范围，比如partitionColumn为ID，ID的范围为[1,1000]，lowerBound、upperBound分别为1 100，numPartitions为10，那么可以分为10个并行的任务去读，比如第二个分区的认为的sql大概是这样过滤的"where ID>=11 and ID<10"。每个分区的ID的范围如下：0: 1-10 ,1:11-20,2:21-30,3:31；40，4：41-50，5：51-60，6：61-70，7：71-80，8：81-90，9：91-1000，需要注意的是Spark会把lowerBound、upperBound范围的数据按照ID等比例（按ID区间）分到每个分区上，而小于lowerBound的部分被分到了个分区，大于upperBound的部分则被分到了后一个分区（大概sql就是个分区的过滤条件没有下限，后一个没有上限~），大家可以自己写程序测试下，这里就不贴程序了。
另numPartitions应该小于等于upperBound-lowerBound
相关文章