关于Coalesce降低Spark JDBC读取并行度的技术咨询

哈哈，这个坑我去年做MySQL到HDFS的批量迁移时踩过！当时排查了半天才发现是Spark的优化器“好心办坏事”——把coalesce的分区调整直接下推到了JDBC读取阶段，完全浪费了我们设置的高并行度参数。

问题到底出在哪？ #

核心就是Spark的惰性求值+Catalyst优化器的算子下推逻辑：

• 你设置numPartitions=42读取MySQL，本来Spark应该启动42个并行任务去拉取数据，效率很高
• 但coalesce(6)是个窄依赖操作（不需要 shuffle，只是合并相邻分区），优化器觉得“既然最后要6个分区，那不如直接用6个分区读数据，省得后面合并”
• 结果就是Spark直接忽略了你设置的numPartitions=42，只用6个分区去读MySQL，读取速度直接暴跌好几倍

解决方案（按推荐优先级排序） #

1. 把coalesce换成repartition（最简单粗暴有效）

直接把df.coalesce(6)改成df.repartition(6)就行。

• 为啥有用？因为repartition是宽依赖操作，会触发shuffle，Catalyst优化器没办法把它下推到JDBC读取阶段
• 效果：读取阶段依然用42个并行任务拉取数据，之后再通过shuffle合并成6个分区写入HDFS
• 小提醒：如果数据量特别大，shuffle会有一定开销，但比起读取速度的提升，这个代价几乎可以忽略；要是实在不想shuffle，看下面的方案

2. 插入“屏障操作”打断优化链（无shuffle）

如果不想触发shuffle，可以在JDBC读取之后、coalesce之前，加个没什么实际作用但能打断优化链的操作，比如：

// Scala示例：加个临时列再删掉，让优化器认为前后是两个独立阶段
val rawDf = spark.read.jdbc(url, table, props).option("numPartitions", 42).load()
// 插入屏障操作
val barrierDf = rawDf.withColumn("__dummy_temp", lit(1)).drop("__dummy_temp")
val finalDf = barrierDf.coalesce(6)

或者用RDD转一圈的方式：

val rawDf = spark.read.jdbc(url, table, props).option("numPartitions", 42).load()
val barrierDf = rawDf.rdd.map(row => row).toDF(rawDf.schema: _*)
val finalDf = barrierDf.coalesce(6)

• 原理：这些操作会让Catalyst优化器无法把coalesce和JDBC扫描合并，读取阶段依然用42个分区，之后再合并成6个
• 优势：完全没有shuffle开销，性能最优

3. 禁用特定优化规则（全局配置）

要是不想改代码，也可以通过Spark配置直接关掉导致这个问题的优化规则：

# 提交作业时加这个参数
--conf spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.CoalesceBeforeScan

或者在代码里设置：

spark.conf.set("spark.sql.optimizer.excludedRules", "org.apache.spark.sql.catalyst.optimizer.CoalesceBeforeScan")

• 原理：CoalesceBeforeScan就是那个把coalesce下推到扫描阶段的优化规则，禁用它就不会出现这个问题了
• 注意：这是全局设置，会影响整个作业的优化逻辑，要是其他地方需要这个优化，谨慎使用

怎么验证是否生效？ #

你可以打印DataFrame的逻辑计划看看：

finalDf.explain(true)

如果看到JDBC扫描阶段的分区数是42，coalesce在后面的阶段，就说明生效了；要是扫描阶段的分区数还是6，那说明优化还在搞事情。

内容的提问来源于stack exchange，提问作者y2k-shubham