PySpark DataFrame经转换后按PartitionBy写入Parquet速度极慢

1. 优先排查数据倾斜问题 #

groupby、pivot操作极易引发数据分布不均，这是大集群下任务耗时过长的核心诱因：

• 执行df.groupBy('year').count().show()，查看各year分区的数据量差异，若某几个year的记录数是其他分区的几十倍甚至上百倍，说明存在严重倾斜。
• 打开Spark UI的Storage标签，检查每个分区的大小，若单个分区远大于128M（默认分区大小阈值），会直接导致单个task处理时间过长。
• 倾斜解决办法：
  • 若pivot时的列基数过高，可拆分任务：按year拆分DataFrame，每个year单独执行pivot后再合并。
  • 对倾斜的year分区单独处理：将大分区拆分为更小的子分区，再与其他小分区合并后写入。

2. 调整分区策略适配partitionBy('year') #

你当前的repartition(320)未结合year列，会导致同一year的数据分散在多个分区中，写入Parquet时需要跨分区合并到对应year目录，反而增加IO开销：

• 改用按year列+合理分区数重分区：先统计year的唯一值数量n，执行df.repartition('year', 320 // n)，让每个year分配到等量的分区（比如10个year的话，每个year分32个分区），这样每个task仅处理单个year的部分数据，写入时直接对应目标目录，避免额外IO消耗。

3. 优化Spark核心配置 #

针对320核的集群，调整以下关键参数：

• spark.sql.shuffle.partitions：默认值200，改为320或640（与集群核数匹配或翻倍），groupby、pivot都会触发shuffle，足够的shuffle分区能避免单个task处理过多数据。
• Executor资源分配：每个节点32核，建议设置spark.executor.cores=8，spark.executor.instances=40（10节点×4），spark.executor.memory=32G（根据节点总内存调整，保证每个executor有足够内存减少GC频率）。
• IO优化：
  • 保持spark.sql.parquet.compression.codec=snappy（默认配置），兼顾压缩比与读写速度。
  • 调整spark.sql.files.maxPartitionBytes：若小文件过多，调至64M；若为大文件，调至256M，让每个分区大小更合理。

4. 精简转换逻辑的冗余 #

回顾你的groupby、pivot操作：

• 是否可在聚合前过滤无效行/冗余列？比如先清理无效数据、删除不必要的列，减少中间数据量。
• pivot操作是否需要生成全量列？若某些列的空值占比极高，可过滤这些列后再执行写入。

5. 借助Spark UI定位瓶颈 #

通过Spark UI的Stages标签分析：

• 查看task执行时间分布，是否存在长尾task（大部分task几秒完成，少数几个耗时超几十分钟），这类task就是性能瓶颈，对应的数据分区需要拆分或优化。
• 查看Shuffle Read/Write的总数据量，若shuffle数据量远超原始数据，说明聚合逻辑有优化空间。

内容的提问来源于stack exchange，提问作者Raja Sabarish PV