PySpark保存Parquet文件时的报错与分区调整问题

问题根源拆解 #

• 40万+分区完全超出合理范围，每个分区数据量极小，直接导致Spark调度Task的开销爆炸：大量Task的元数据累加引发WARN DAGScheduler: Broadcasting large task binary警告，终端被scala.collection.Iterator$$anon$11.hasNext日志刷屏，是因为大量小分区的迭代操作频繁触发底层Scala迭代器的调用。
• 调整分区时执行时间过长，大概率是选错了分区调整方法（比如用repartition触发全量shuffle），或者调整时机太晚（数据已经膨胀后再处理）。

核心解决思路 #

1. 先定位分区膨胀的源头 #

从数据处理全链路排查：

• 是否读取了大量小Parquet文件？小文件会直接对应大量分区。
• 是否是中间操作（比如flatMap、explode）生成海量数据行，导致分区自动分裂？
• 是否是宽依赖操作（比如join、groupBy）后，继承了上游过多的分区数？

2. 针对性压制分区膨胀 #

• 读取小文件场景：读取时通过参数合并小文件，示例代码：

  df = spark.read.option("mergeSchema", "true") \
                .option("recursiveFileLookup", "true") \
                .parquet("/path/to/data")
  

  也可以先通过文件系统命令合并小文件后再读取。
• 中间操作导致分区暴涨：在该操作后立即用coalesce收缩分区，避免分区数持续累积。比如flatMap后直接追加coalesce(200)，不要等到最后才处理。

分区优化实战方案 #

1. 合理计算最优分区数 #

Spark最优分区大小通常在128MB-256MB之间，按此标准估算：

最优分区数 = 总数据量 ÷ 单分区目标大小（比如256MB）

单节点场景下，同时兼顾CPU核心数，分区数建议不超过核心数的3-4倍，避免调度过载。比如8核CPU，分区数控制在24-32之间即可。

2. 选对分区调整方法 #

• coalesce优先：仅用于减少分区，属于窄依赖操作，不会触发shuffle，性能极高。你的临时方案coalesce(200)是正确的，但要提前到数据膨胀前执行，减少后续操作的开销。
• repartition慎用：用于增加分区或重新均匀分布数据，会触发全量shuffle，性能开销大，仅在数据倾斜、分区分布极不均匀时使用。

3. 配置层面优化 #

• 设置spark.sql.shuffle.partitions：默认200，根据集群/单节点资源调整，比如8核CPU设为16-24，避免宽依赖操作后生成过多分区。
• 设置spark.default.parallelism：控制RDD的默认分区数，建议设为CPU核心数的2-3倍。

4. Parquet保存时的额外优化 #

• 避免用高基数字段partitionBy：比如按用户ID分区会生成百万级小文件，优先用日期、地区等低基数业务字段。
• 开启压缩：保存时添加option("compression", "snappy")，减少文件大小，提升读写效率。
• 批量写入：如果是循环写入数据，尽量攒够一定量再合并写入，避免生成大量小文件。

5. 数据倾斜处理（若存在） #

如果调整分区后仍有性能瓶颈，检查是否存在数据倾斜：

• 查看每个分区的数据量，定位数据量异常大的分区。
• 用加盐法拆分大分区：比如对倾斜字段添加随机后缀，拆分后再聚合；或者将大表拆分为小表分别处理。

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