Spark JDBC 调优

val sparkSession = SparkSession.builder.appName("jdbc learn").getOrCreate()
val reader:DataFrameReader = sparkSession.read.format("jdbc")
                                   .option("url", "jdbc:oracle:thin:@<host>:<port>:<SID> ")
                                   // oracle 表自带整数列 rownum
                                   .option("dbtable", "(SELECT a.*, rownum as rn FROM tablename a)")
                                   .option("user", "user")
                                   .option("password", "password")
                                   .option("fetchsize", 500)
                                   // 分配分区数(task)尽量是cores的倍数
                                   .option("numPartitions", 32)
                                   // 指定数字类型的列
                                   .option("partitionColumn", "rn")
                                   // 分区上下界
                                   .option("lowerBound", "1")
                                   .option("upperBound", "5000000")

package org.apache.spark.sql.execution.datasources.jdbc

/**
 * Instructions on how to partition the table among workers.
 */
private[sql] case class JDBCPartitioningInfo(
    column: String,
    lowerBound: Long,
    upperBound: Long,
    numPartitions: Int)

private[sql] object JDBCRelation extends Logging {
  /**
   * Given a partitioning schematic (a column of integral type, a number of
   * partitions, and upper and lower bounds on the column's value), generate
   * WHERE clauses for each partition so that each row in the table appears
   * exactly once.  The parameters minValue and maxValue are advisory in that
   * incorrect values may cause the partitioning to be poor, but no data
   * will fail to be represented.
   *
   * Null value predicate is added to the first partition where clause to include
   * the rows with null value for the partitions column.
   *
   * @param partitioning partition information to generate the where clause for each partition
   * @return an array of partitions with where clause for each partition
   */
  def columnPartition(partitioning: JDBCPartitioningInfo): Array[Partition] = {
    if (partitioning == null || partitioning.numPartitions <= 1 ||
      partitioning.lowerBound == partitioning.upperBound) {
      return Array[Partition](JDBCPartition(null, 0))
    }

    val lowerBound = partitioning.lowerBound
    val upperBound = partitioning.upperBound
    require (lowerBound <= upperBound,
      "Operation not allowed: the lower bound of partitioning column is larger than the upper " +
      s"bound. Lower bound: $lowerBound; Upper bound: $upperBound")

    /**
     * 笔者注：
     * 从这里可以看出`upperBound`和`lowerBound`的差如果小于你指定的`numPartitions`
     * 那么实际的分区数其实是 upperBound-lowerBound
     */
    val numPartitions =
      if ((upperBound - lowerBound) >= partitioning.numPartitions || /* check for overflow */
          (upperBound - lowerBound) < 0) {
        partitioning.numPartitions
      } else {
        logWarning("The number of partitions is reduced because the specified number of " +
          "partitions is less than the difference between upper bound and lower bound. " +
          s"Updated number of partitions: ${upperBound - lowerBound}; Input number of " +
          s"partitions: ${partitioning.numPartitions}; Lower bound: $lowerBound; " +
          s"Upper bound: $upperBound.")
        upperBound - lowerBound
      }
    // Overflow and silliness can happen if you subtract then divide.
    // Here we get a little roundoff, but that's (hopefully) OK.
    // 笔者注：分区的跨度
    val stride: Long = upperBound / numPartitions - lowerBound / numPartitions
    val column = partitioning.column
    var i: Int = 0
    var currentValue: Long = lowerBound
    var ans = new ArrayBuffer[Partition]()
    /**
     * 笔者注：
     * 假设指定`partitionColumn = 32`，且`partitionColumn - lowerBound >= partitionColumn`。
     * 对于 partition_i，构造条件语句如下： 
     * partition_0 : whereClause = $partitionColumn < $lowerBound + $stride or $partitionColumn is null
     * partition_1~30 : whereClause = $partitionColumn >= $lowerBound + $stride * i AND $partitionColumn < $lowerBound + $stride * (i + 1)
     * partition_31 : whereClause = $partitionColumn >= $lowerBound + $stride * i
     * 注意：$lowerBound + $stride * 31 并不一定等于 $upperBound - $stride
     */
    while (i < numPartitions) {
      // 笔者注：构造分区下界条件语句，若是第一个分区(partition0)，下界条件为null
      val lBound = if (i != 0) s"$column >= $currentValue" else null
      currentValue += stride
      // 笔者注：构造分区上界条件语句，若是最后一个分区，上界条件为null
      val uBound = if (i != numPartitions - 1) s"$column < $currentValue" else null
      val whereClause =
        if (uBound == null) {
          lBound
        } else if (lBound == null) {
          s"$uBound or $column is null"
        } else {
          s"$lBound AND $uBound"
        }
      ans += JDBCPartition(whereClause, i)
      i = i + 1
    }
    ans.toArray
  }
}

// 为了不丢失数据，向上取整，将数据分成32份
val stride = Math.ceil(1384288 / 32)

val tableName = "TABLE"

// 创建32个task
val registerDF = Range(0, 32)
  .map {
    index =>
      sparkSession
        .read
        .format("jdbc")
        .option("url", jdbcProps.getProperty("url"))
        .option("dbtable", s"(SELECT * FROM (SELECT a.*, rownum as rn FROM $tableName a) b WHERE b.rn > ${stride * index} AND b.rn <= ${stride * (index + 1)})")
        .option("user", jdbcProps.getProperty("user"))
        .option("password", jdbcProps.getProperty("password"))
        .option("fetchsize", 500)
        .load()
  }
  .reduce((resultDF1, resultDF2) => resultDF1.union(resultDF2))