Spark JDBC读取海量数据时如何设置lowerBound与upperBound高效分区

关于是否必须使用真实min/max作为上下界 #

不需要使用真实的min/max，但必须满足一个核心前提：你设置的lowerBound到upperBound的区间，必须完全覆盖本次查询所有数据的partitionColumn取值。只要满足这个条件，即使上下界比真实值范围大，也不会丢数据，只是会生成部分没有数据的空task，这类task执行速度极快，对整体任务耗时影响可以忽略。
如果上下界范围小于真实取值范围，落在区间外的数据会被直接丢弃，这是必须避免的问题。

无需提前查询min/max的落地方案 #

方案1：直接使用日期列作为分区列（最优） #

你当前已经按年月做过滤，且表中存在可用的日期列，完全可以直接用日期列/时间戳作为partitionColumn，不需要找整数类型的分区列：

• lowerBound直接取当前读取月份的第一天，比如2016-01-01 00:00:00
• upperBound直接取下一个月的第一天，比如2016-02-01 00:00:00
• numPartitions可以根据当月天数设置为32，或者根据你需要的并发度设置为16/64都可

Spark会自动将时间区间按numPartitions拆分成对等的子区间，每个分区拉取对应时间范围的数据，天然均匀无倾斜，所有参数都可以通过日期计算直接得到，完全不需要查询数据库。
如果担心JDBC驱动对日期类型的兼容性，也可以提前把日期转成13位毫秒时间戳整数，用时间戳作为分区列，逻辑完全一致。
示例代码：

val startTime = YearMonth.of(year.toInt, month).atDay(1).atStartOfDay(ZoneId.systemDefault()).toInstant.toEpochMilli
val endTime = YearMonth.of(year.toInt, month).plusMonths(1).atDay(1).atStartOfDay(ZoneId.systemDefault()).toInstant.toEpochMilli
val query = s"(select *, unix_timestamp(date_time)*1000 as ts from table where date_column >= '${YearMonth.of(year.toInt, month).atDay(1).toString}' and date_time <= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as datadf"
val dataframe = spark.read.format("jdbc")
          .option("url", url)
          .option("driver", driver)
          .option("user", user)
          .option("password", password)
          .option("dbtable", query)
          .option("partitionColumn", "ts")
          .option("lowerBound", startTime)
          .option("upperBound", endTime)
          .option("numPartitions", 32)
          .option("fetchsize", 15000)
          .load()

方案2：使用覆盖全范围的固定上下界 #

如果必须使用现有的整数类型作为partitionColumn，可以直接设置足够大的上下界区间，覆盖该列所有可能的取值即可：
比如你可以根据数据分布预估该整数列2016-2021年的取值范围在0到3000亿之间，直接固定设置：

.option("lowerBound", 0)
.option("upperBound", 300000000000L)
.option("numPartitions", 200)

由于你已经在dbtable的查询中加了日期过滤条件，数据库会先过滤出符合日期条件的数据，再按Spark拆分的整数区间拉取数据，不会扫全表，空分区的执行开销远低于提前查询min/max的开销。

方案3：优化现有min/max查询 #

如果还是想使用真实min/max，你当前的min/max查询慢大概率是没有走日期列索引，可以给查询加索引提示强制走索引：
比如MySQL可以改写为：

(select min(partitionColumn) as mn, max(partitionColumn) as mx from tablename force index(date_column_idx) where date_column >= 'xxx' and date_time <= 'xxx') as boundsDF

Oracle可以加hint：

(select /*+ INDEX(tablename date_column_idx) */ min(partitionColumn) as mn, max(partitionColumn) as mx from tablename where date_column >= 'xxx' and date_time <= 'xxx') as boundsDF

走索引的情况下min/max查询基本可以秒返回，不会成为性能瓶颈。

内容的提问来源于stack exchange，提问作者Metadata