Spark重分区后数据分布均匀仍有少数任务运行过慢问题咨询

你调整的repartitionByRange逻辑仅作用于Spark作业运行期的内存态DataFrame分区，和最终写入阶段的分区逻辑完全独立。写入时调用的partitionBy(col1, col2, col3)会触发一次shuffle，将所有归属同一(col1,col2,col3)业务分区的数据收拢到同一个写任务处理，只要某组业务分区对应的数据量本身极大，无论前面的内存分区多均匀，写阶段都会出现单任务过载的问题。

• 方案1：加盐打散业务分区写入 #

  在不影响业务读取逻辑的前提下，给业务分区增加随机盐后缀，将单业务分区的写入压力拆分到多个任务并行处理：

  import org.apache.spark.sql.functions.rand
  // 随机盐取值范围0-9，可根据单业务分区数据量级调整拆分份数
  val dfWithSalt = dfIn.withColumn("salt", (rand() * 10).cast("int"))
  // 按业务分区+盐重分区，减少写阶段shuffle开销
  val dfRepartitioned = dfWithSalt.repartition(col1, col2, col3, col("salt"))
  // 写入时将盐加入分区列，单业务分区下会生成多份带salt后缀的子分区
  dfRepartitioned.write.mode(SaveMode.Append)
    .partitionBy(col1, col2, col3, "salt")
    .parquet(PATH_OUT)
  

  后续读取数据时，Spark会自动识别同业务分区下的所有salt子分区，指定col1/col2/col3过滤条件时无需额外处理salt字段，不影响原有查询逻辑。如果对小文件数量有要求，可后续在低峰期定期合并同业务分区下的文件。

• 方案2：调整写入阶段配置 #

  如果不能修改输出分区结构，可通过调整Spark参数降低写任务压力：

  • 开启自适应查询(AQE)倾斜优化，自动拆分写阶段的大shuffle分区：

    spark.sql.adaptive.enabled=true
    spark.sql.adaptive.skewJoin.enabled=true
    spark.sql.adaptive.skewPartitionThresholdInBytes=128MB // 根据实际场景调整倾斜判定阈值
    

  • 关闭单输出分区的文件合并限制，提升并发写能力：

    spark.sql.sources.maxConcurrentWrites=10 // 与拆分的盐份数匹配即可
    spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2
    

  • 优化序列化与压缩配置，降低单任务CPU开销：

    spark.serializer=org.apache.spark.serializer.KryoSerializer
    spark.sql.parquet.compression.codec=snappy
    

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