Apache Spark DataFrame写入Parquet时repartition不生效问题求助

我之前也踩过一模一样的坑！先帮你拆解下背后的原因，再给你几个靠谱的解决方案：

为什么会出现这种矛盾现象？ #

首先得搞清楚repartition和coalesce的核心差异，以及Spark优化器的“小心思”：

• coalesce减少分区：默认不会触发shuffle，它只是把现有分区的数据合并到更少的分区里，相当于在数据读取完成后做“合并操作”，Spark优化器没法跳过这个步骤，所以写入时自然会用合并后的分区数。
• repartition（无论增减分区）：会触发shuffle，但Spark的Catalyst优化器为了提升性能，可能会把repartition操作和前面的数据源读取操作做“融合优化”——比如如果你的初始DataFrame是从有4000个分区的Parquet文件读来的，优化器会觉得“直接读原分区写入更高效”，直接跳过了repartition的shuffle步骤，导致内存里的分区数显示是20，但实际写入还是用了原数据源的4000个分区。

而当你要增加分区时（比如从8到100），如果优化器同样跳过了repartition的shuffle，那自然还是用原有的8个分区写入。

靠谱的解决方案 #

1. 用缓存打断优化器的“小聪明” #

最简单的办法就是在repartition之后加个cache()或者persist()，让Spark先把重分区后的DataFrame缓存到内存里，再执行写入操作——这样优化器就没法把repartition和读取操作合并了：

// 减少分区的情况
var df_new = df.repartition(20).cache()
// 先触发缓存（比如查看分区数或执行count）
println(df_new.rdd.partitions.size) // 确认是20
df_new.write.parquet("test.parquet")

// 增加分区的情况
var df_new = df.repartition(100).cache()
df_new.count() // 触发缓存加载
df_new.write.parquet("test_increased.parquet")

如果数据量太大，内存缓存不下，可以用checkpoint()（需要先设置checkpoint目录）：

spark.sparkContext.setCheckpointDir("/tmp/spark_checkpoint")
var df_new = df.repartition(100).checkpoint()
df_new.write.parquet("test_increased.parquet")

2. 指定分区键触发强制shuffle #

如果你的数据有合适的分区列，比如用户ID、日期之类的，可以在repartition时指定这个列，这样Spark一定会触发shuffle操作，优化器没法跳过：

// 按user_id列分区，同时设置总分区数为100
var df_new = df.repartition(100, $"user_id")
df_new.write.parquet("test_partitioned.parquet")

这种方式不仅能保证分区数正确，还能让相同键的数据落到同一个分区里，对后续查询也更友好。

3. 用coalesce强制shuffle（适合增减分区） #

其实coalesce也可以用来增加分区，只要设置shuffle=true——这时候它的效果和repartition完全一样，但写法上更明确：

// 从8个分区增加到100个，强制触发shuffle
var df_new = df.coalesce(100, shuffle = true)
df_new.write.parquet("test_coalesce_increased.parquet")

4. 临时关闭优化规则（不推荐，仅排查用） #

如果上面的方法都不行，可以临时关闭导致repartition被跳过的优化规则，比如列裁剪、谓词下推等：

// 关闭指定的优化规则
spark.conf.set("spark.sql.optimizer.excludedRules", "org.apache.spark.sql.catalyst.optimizer.PushDownPredicate,org.apache.spark.sql.catalyst.optimizer.ColumnPruning")
// 再执行repartition和写入
var df_new = df.repartition(100)
df_new.write.parquet("test_no_optim.parquet")

注意：这种方法会影响Spark整体查询性能，只适合临时排查问题，生产环境别用。

总结一下 #

• 减少分区：用coalesce更高效（不shuffle），或者repartition加缓存；
• 增加分区：必须用repartition（或coalesce(..., shuffle=true)），并且通过缓存、指定分区键等方式强制触发shuffle，避免优化器跳过操作。

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