Spark应用中应当在什么时候对DataFrame执行repartition操作？

核心结论 #

你的思路具备合理性，但手动repartition不是所有场景的通用解法，需要先根因定位问题再决定是否操作，Spark不会默认在读取parquet后按后续join键做预分区。

问题现象根因排查 #

1. 多executor未分配输入任务的原因 #

读取parquet时的初始DataFrame分区数由以下规则决定，和后续join逻辑无关：

• Spark默认按spark.sql.files.maxPartitionBytes参数（默认值128MB）切分输入文件，单个文件小于该阈值时不会被拆分
• 若你的parquet文件数量少、或大量小文件合并后总分区数远小于executor总核心数，就会出现部分executor无任务可跑的情况

2. 理论均匀的join键仍出现shuffle倾斜的可能原因 #

Spark执行普通join时默认会自动对两个输入DataFrame按join键做shuffle分区，分区数由spark.sql.shuffle.partitions参数控制（默认值200），如果仍然出现倾斜，优先排查以下问题：

• 键实际分布不符合预期：优先执行统计命令验证分布：df.groupBy("join_id").count().orderBy(col("count").desc()).show(20)，确认是否存在未预期的热点键、大量空值ID
• shuffle分区数不足：数百GB规模的作业默认200个shuffle分区会导致单分区数据量过大，容易出现分配不均
• AQE配置异常：Spark 3.0+默认开启的自适应查询执行（AQE）会自动优化倾斜，若你关闭了AQE、或倾斜检测阈值设置过高也会导致倾斜未被自动处理
• 存在小表join大表场景：如果其中一个join的DataFrame内存可放下，未使用broadcast join会导致不必要的shuffle，也容易出现倾斜表象

手动repartition的适用场景 #

满足以下任一条件时，建议在读取parquet后按join键执行手动repartition：

• 你需要对同一个DataFrame按该ID键执行多次join、聚合操作：预分区后后续操作无需重复shuffle，可大幅降低整体shuffle数据量
• 读入阶段初始分区数远小于executor可用核心数，读阶段就出现大量executor闲置：手动repartition可打散数据，提升资源利用率
• 示例代码：val repartitionedDf = rawDf.repartition(400, col("join_id"))，可根据数据规模调整分区数

无需手动repartition的场景 #

如果仅对该DataFrame执行一次join操作，后续不再使用该ID键做关联/聚合，无需提前repartition，否则会多执行一次不必要的shuffle，反而浪费资源，直接优化join阶段参数即可。

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