如何提升Spark中基于列表达式的重分区性能？

问题背景 #

你需要将月度数据按dailyDt分区，每个日期生成20个大小均匀的文件，但当前用rand()加盐的方案耗时过长。以下是高效的优化手段：

1. 用确定性哈希分桶替代随机加盐（最优方案） #

rand()的随机计算和shuffle时的不确定性是性能瓶颈，改用确定性的哈希取模可以在保证文件均匀的同时大幅提升速度：

import org.apache.spark.sql.functions.{monotonically_increasing_id, floor}

df.withColumn("bucket", floor(monotonically_increasing_id() % 20))
  .repartition(20, $"dailyDt", $"bucket")
  .drop("bucket")
  .write
  .partitionBy("dailyDt")
  .mode(Overwrite)
  .parquet("/path..")

• 逻辑：通过monotonically_increasing_id()生成全局递增ID，对20取模得到0-19的桶编号，确保每个dailyDt内的数据被均匀分配到20个分区
• 优势：计算效率远高于随机数，shuffle过程数据分布稳定，避免不必要的IO开销

2. 调优Spark Shuffle参数 #

针对大数据量shuffle场景，调整以下参数减少磁盘IO、提升内存利用率：

// 设置shuffle分区数（建议为集群核心数的2-3倍，或等于目标总文件数：20*30=600）
spark.conf.set("spark.sql.shuffle.partitions", 600)
// 增大shuffle文件缓冲区，减少磁盘写入次数
spark.conf.set("spark.shuffle.file.buffer", "64k")
// 扩大Reducer端数据拉取缓冲区，降低磁盘IO频率
spark.conf.set("spark.reducer.maxSizeInFlight", "96m")
// 调整shuffle内存占比，避免内存溢出同时提升效率
spark.conf.set("spark.shuffle.memoryFraction", 0.3)

3. 预分区后直接写入（减少二次shuffle） #

如果数据本身没有合理分区，可以先完成预分区操作，再执行写入，避免写入阶段的额外shuffle：

// 预分区：按dailyDt和桶列划分，总分区数匹配目标文件数
val prePartitionedDf = df.repartition(600, $"dailyDt", floor(monotonically_increasing_id() % 20))
prePartitionedDf.write
  .partitionBy("dailyDt")
  .mode(Overwrite)
  .parquet("/path..")

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