如何控制Spark合并HDFS小Parquet文件输出大小并减少repartition shuffle

原方案问题根源 #

• 输出文件总大小超过输入：普通repartition采用哈希规则打散数据，破坏了原Parquet文件同列数据连续存储的结构，导致压缩算法压缩率大幅下降；若写入时未指定与输入一致的压缩配置，也会引发文件膨胀。
• 大量shuffle开销：repartition属于宽依赖操作，全量数据需要通过网络打散分发到不同分区，数据量越大开销越高。

可行优化方案 #

1. 优先用coalesce替代repartition #

合并小文件属于减少分区数的场景，coalesce是窄依赖操作，不会产生shuffle，仅在本地合并分区即可，执行效率远高于repartition。注意不要一次性将分区数降得过低，避免单节点负载过高。

2. 用Spark原生参数自动控制文件大小，无需手动计算Magic Number #

Spark原生提供了精准控制输出文件大小的配置，适配性远高于手动计算的分区数：

• 开启自适应查询执行（AQE，Spark 3.0+支持），自动合并小分区：

spark.conf.set("spark.sql.adaptive.enabled", "true")
// 单分区目标大小设为512MB
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "536870912")

• 配置单文件最大记录数，避免单个文件过大：可根据单条记录的平均大小反推数值，比如单条记录平均大小为1KB时，512MB对应524288条：

spark.conf.set("spark.sql.files.maxRecordsPerFile", 524288)

• 写入时指定与输入一致的压缩算法，保证压缩率不会下降：比如原文件用Snappy压缩，就显式指定压缩配置。

3. 低版本Spark兼容方案（Spark 2.x无AQE） #

如果使用不支持AQE的Spark版本，可以用repartitionByRange替代普通repartition，选择基数较高的列作为分区键，排序后的列式存储压缩率更高，既可以避免文件膨胀，还能提升后续按该列查询的效率。

完整优化代码示例（Spark 3.x） #

# 配置参数
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "536870912")
spark.conf.set("spark.sql.files.maxRecordsPerFile", 524288)
# 开启列存写入优化，进一步提升压缩率
spark.conf.set("spark.sql.parquet.columnarWriter.enabled", "true")

# 读取原文件
df = spark.read.parquet("/input/hdfs/path")

# 若原分区数远大于预期，先做coalesce减少分区
if df.rdd.getNumPartitions() > 200:
    df = df.coalesce(200)

# 写入文件，指定压缩算法
df.write.option("compression", "snappy") \
  .mode("overwrite") \
  .save("/output/hdfs/path")

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