Spark 3.2.1中AQE大幅缩减分区致单Executor负载过高，如何解决？

原因分析 #

1. AQE自动合并分区逻辑触发
   Spark 3.2.1的自适应执行（AQE）默认会根据shuffle后的数据大小合并小分区，核心参数spark.sql.adaptive.advisoryPartitionSizeInBytes默认值为64MB。如果Join后的shuffle输出中，大量分区数据量远小于该阈值，AQE会自动将这些小分区合并，最终导致分区数骤降（比如你这里从900降到8）。而其中一个分区包含90GB数据，说明存在严重的数据倾斜——Join的key分布极不均匀，绝大多数数据集中在单个key对应的分区里，AQE无法拆分这个超大分区，只能让单个executor处理。

2. 关闭合并分区仍无效的原因
   即使你关闭了spark.sql.adaptive.coalescePartitions.enabled，如果数据倾斜本身存在，shuffle后的分区数据分布依然会严重不均，大分区的问题不会消失。另外，Spark 3.2.1的AQE还包含其他优化逻辑（比如动态调整shuffle分区、倾斜Join处理），如果这些逻辑没被完全关闭，仍可能影响分区数量。

3. 与row_number()无关的原因
   移除窗口函数后问题依旧，说明负载集中的根源不在窗口函数的分区逻辑，而是Join阶段的数据分布问题——窗口函数只是基于上游的分区进行计算，不会改变上游的分区倾斜情况。

优化方案 #

1. 针对性解决数据倾斜 #

• 加盐打散skew key：对倾斜的Join key添加随机后缀（比如0~9的随机数），将单个大分区拆分为多个小分区进行Join，之后再合并结果。示例代码：

  // 对倾斜表的key加盐
  val skewedDF = df.withColumn("skew_key_salt", concat(col("skew_key"), lit("_"), floor(rand() * 10)))
  // 对另一表的key生成对应后缀的所有可能值，再执行Join
  val normalDF = df2.withColumn("skew_key_salt", explode(array((0 to 9).map(lit(_)))))
                     .withColumn("skew_key_salt", concat(col("skew_key"), lit("_"), col("skew_key_salt")))
  val joinedDF = skewedDF.join(normalDF, Seq("skew_key_salt"), "inner")
                         .drop("skew_key_salt")
  

• 启用AQE倾斜Join处理：确保spark.sql.adaptive.skewJoin.enabled=true（Spark 3.2.1默认开启），并调整倾斜阈值参数：

  spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=1024mb  # 调大阈值，确保大分区被检测到
  spark.sql.adaptive.skewJoin.skewedPartitionFactor=5  # 分区数据量超过平均5倍即判定为倾斜
  

  AQE会自动将倾斜的key单独拆分出来，用broadcast join的方式处理，避免大分区出现。

2. 调整AQE合并分区参数，避免分区过度减少 #

• 设置最小分区数，防止AQE把分区合并得太少：

  spark.sql.adaptive.coalescePartitions.minPartitionNum=200  # 根据集群资源设置合适值
  

• 调大分区大小阈值，减少合并行为：

  spark.sql.adaptive.advisoryPartitionSizeInBytes=1024mb  # 让AQE只合并小于1GB的分区
  

3. 手动干预分区，打散负载 #

在Join操作之后、汇总操作之前，手动执行repartition或repartitionByRange，强制打散数据：

val joinedDF = df1.join(df2, "join_key")
val repartitionedDF = joinedDF.repartition(900, col("group_by_key"))  # 按汇总key重新分区

这样后续的汇总计算会分散到多个executor执行，避免单个节点负载过高。

4. 关闭不必要的AQE优化（仅在必要时） #

如果不需要AQE的其他优化，可以完全关闭AQE，强制使用固定的shuffle分区数：

spark.sql.adaptive.enabled=false
spark.sql.shuffle.partitions=900

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