本文旨在介绍 Spark 通过JDBC读取数据库数据的四种API
调用前准备
对于不同的数据库，需要在spark的环境中添加对应的driver包，如：
Oracle：
export SPARK_CLASSPATH=$SPARK_HOME/ojdbcxx.jar (具体版本号自定)
MySQL：
export SPARK_CLASSPATH=/path/mysql-connector-java-5.x.x.jar (具体版本号自定)
其他数据库可以参考官方文档：官方JDBC配置文档
四种API简介
1.单分区模式
函数：
def jdbc(url: String, table: String, properties: Properties): DataFrame
使用示例：
val url = "jdbc:mysql://mysqlHost:3306/database"
val tableName = "table"
// 设置连接用户&密码
val prop = new java.util.Properties
prop.setProperty("user","username")
prop.setProperty("password","pwd")
// 取得该表数据
val jdbcDF = sqlContext.read.jdbc(url,tableName,prop)
// 一些操作
....
从入参可以看出，只需要传入JDBC URL、表名及对应的账号密码Properties即可。但是计算此DF的分区数后发现，这种不负责任的写法，并发数是1
jdbcDF.rdd.partitions.size=1
操作大数据集时，spark对MySQL的查询语句等同于可怕的：select * from table; ,而单个分区会把数据都集中在一个executor，当遇到较大数据集时，都会产生不合理的资源占用：MySQL可能hang住，spark可能会OOM，所以不推荐生产环境使用；
2.指定Long型column字段的分区模式
函数：
def jdbc(
url: String,
table: String,
columnName: String,
lowerBound: Long,
upperBound: Long,
numPartitions: Int,
connectionProperties: Properties): DataFrame
使用id做分片字段的示例：
val url = "jdbc:mysql://mysqlHost:3306/database"
val tableName = "table"
val columnName = "id"
val lowerBound = getMinId()
val upperBound = getMaxId()
val numPartitions = 200
// 设置连接用户&密码
val prop = new java.util.Properties
prop.setProperty("user","username")
prop.setProperty("password","pwd")
// 取得该表数据
val jdbcDF = sqlContext.read.jdbc(url,tableName, columnName, lowerBound, upperBound,numPartitions,prop)
// 一些操作
....
从入参可以看出，通过指定 id 这个数字型的column作为分片键，并设置最大最小值和指定的分区数，可以对数据库的数据进行并发读取。是不是numPartitions传入多少，分区数就一定是多少呢？其实不然，通过对源码的分析可知：
if upperBound-lowerBound >= numPartitions:
jdbcDF.rdd.partitions.size = numPartitions
else
jdbcDF.rdd.partitions.size = upperBound-lowerBound
拉取数据时，spark会按numPartitions均分最大最小ID，然后进行并发查询，并最终转换成RDD，例如：
入参为：
lowerBound=1, upperBound=1000, numPartitions=10
对应查询语句组为：
JDBCPartition(id < 101 or id is null,0),
JDBCPartition(id >= 101 AND id < 201,1),
JDBCPartition(id >= 201 AND id < 301,2),
JDBCPartition(id >= 301 AND id < 401,3),
JDBCPartition(id >= 401 AND id < 501,4),
JDBCPartition(id >= 501 AND id < 601,5),
JDBCPartition(id >= 601 AND id < 701,6),
JDBCPartition(id >= 701 AND id < 801,7),
JDBCPartition(id >= 801 AND id < 901,8),
JDBCPartition(id >= 901,9)
建议在使用此方式进行分片时，需要评估好 numPartitions 的个数，防止单片数据过大；同时需要column字段的索引建立情况，防止查询语句出现慢SQL影响取数效率。
如果column的数字是离散型的，为了防止拉取时出现过多空分区，以及不必要的一些数据倾斜，需要使用特殊手段进行处理，具体可以参考Spark JDBC系列--读取优化。
3.高自由度的分区模式
函数：
def jdbc(
url: String,
table: String,
predicates: Array[String],
connectionProperties: Properties): DataFrame
使用给定分区数组的示例：
/**
* 将近90天的数据进行分区读取
* 每一天作为一个分区，例如
* Array(
* "2015-09-17" -> "2015-09-18",
* "2015-09-18" -> "2015-09-19",
* ...)
**/
def getPredicates = {
val cal = Calendar.getInstance()
cal.add(Calendar.DATE, -90)
val array = ArrayBuffer[(String,String)]()
for (i <- 0 until 90) {
val start = new SimpleDateFormat("yyyy-MM-dd").format(cal.getTime())
cal.add(Calendar.DATE, +1)
val end = new SimpleDateFormat("yyyy-MM-dd").format(cal.getTime())
array += start -> end
}
val predicates = array.map {
case (start, end) => s"gmt_create >= '$start' AND gmt_create < '$end'"
}
predicates.toArray
}
val predicates = getPredicates
//链接操作
...
从函数可以看出，分区数组是多个并行的自定义where语句，且分区数为数据size：
jdbcDF.rdd.partitions.size = predicates.size
建议在使用此方式进行分片时，需要评估好 predicates.size 的个数，防止防止单片数据过大；同时需要自定义where语句的查询效率，防止查询语句出现慢SQL影响取数效率。
4.自定义option参数模式
函数示例：
val jdbcDF = sparkSession.sqlContext.read.format("jdbc")
.option("url", url)
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "table")
.option("user", "user")
.option("partitionColumn", "id")
.option("lowerBound", 1)
.option("upperBound", 10000)
.option("fetchsize", 100)
.option("xxx", "xxx")
.load()
从函数可以看出，option模式其实是一种开放接口，spark会根据具体的参数，来决定使用上述三种方式中的某一种。
所有支持的参数具体可以参考官方文档：官方JDBC配置文档 此处附上截图：
结语
JDBC的读取性能受很多条件影响，需要根据不同的数据库，表，索引，数据量，spark集群的executor情况等综合考虑，线上环境的操作，建议进行读写分离，即读备库，写主库。
