AQE / Shuffle 调优实战

难度与前置

难度：中级
前置：会看 Stage / Shuffle Read；能改 spark.sql.adaptive.*。
读完能做什么：打开并验证 AQE 关键能力，按数据量设分区与合并阈值，用 Stage 指标验收。

适用场景

• Spark 3.x 批作业：分区过多/过少、Shuffle 毛刺、轻量倾斜
• 动态数据量（活动日 vs 平日）导致静态 spark.sql.shuffle.partitions=200 失效
• Join / 聚合后小分区爆炸或单分区过大

不适用：

• Spark 2.x 无 AQE（用手工 repartition / 静态调参）
• 严重热点键（AQE 倾斜优化有限，仍要 盐打散 / 两阶段）
• 已 OOM 在 Driver 收集大数据（先看 OOM，不是先开 AQE）

真实场景（一条）

dws_ad_channel_play_1d 任务：平日输入约 80GB，活动日约 400GB。静态 shuffle.partitions=200：

• 平日：大量空/小分区，调度开销高
• 活动日：单任务 Shuffle Read 飙到数 GB，偶发 spill 与长尾

开启 AQE 后：自动合并小分区、局部处理倾斜，活动日 P95 从 45 分钟 → 18 分钟，平日也不再无意义地开 200 个空任务。

怎么做（步骤清单）

1. 打开 AQE 并确认版本能力

# spark-submit / 平台配置示意
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.localShuffleReader.enabled=true

Spark UI 验证：SQL 页出现 AdaptiveSparkPlan，Stage 有「合并后的分区数」变化痕迹。

2. 先设「目标分区大小」，再谈数量

参数	作用	实践起点
spark.sql.adaptive.advisoryPartitionSizeInBytes	合并后目标大小	64MB～128MB
spark.sql.adaptive.coalescePartitions.minPartitionSize	过小则合并	如 16MB～32MB
spark.sql.adaptive.coalescePartitions.initialPartitionNum	初始上限	可高于旧 shuffle.partitions
spark.sql.shuffle.partitions	无 AQE 或初始	保留合理上限作兜底

口诀：按字节调，不按玄学个数调。

3. Shuffle 过重时的检查顺序

1. Spark UI → Stage：Shuffle Read Size、Records、Spill
2. 是否广播可关未关：小表 Join 看 BroadcastHashJoin
3. 是否倾斜：任务耗时/读取量 Pmax ÷ P50 > 3
4. 是否分区过碎：任务数远大于 CPU，且每任务秒级结束

-- 用 EXPLAIN / UI 确认 Join 策略（示意）
EXPLAIN EXTENDED
SELECT /*+ BROADCAST(d) */ f.*, d.settle_entity
FROM fact_spend f
JOIN dim_channel_df d ON f.channel_id = d.channel_id AND f.dt = d.dt
WHERE f.dt = '${biz_date}';

4. AQE 倾斜 Join（轻中度）

spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB

含义：某分区大小超过阈值且远高于中位数时，拆分子任务读。

仍不够时：

• 加盐 / 两阶段聚合（见倾斜专篇）
• 过滤空键、脏键
• 拆开「超热维值」单独路径

5. 与写入、小文件联动

AQE 合并的是 Shuffle 中间分区；最终表文件数还取决于 write 端：

// 写出前按目标文件数整理（避免 AQE 救了计算、写出又打碎）
val targetFiles = Math.max(1, (estimatedBytes / (128L << 20)).toInt)
df.coalesce(targetFiles) // 或 repartition，按是否缩小选择
  .write.mode("overwrite").parquet(path)

详见 小文件治理。

6. 调参实验怎么做才不瞎

1. 固定输入分区与代码版本
2. 只改一类开关（先 coalesce，再 skewJoin）
3. 记录：总耗时、最长 Stage、Spill、峰值内存、输出文件数
4. 活动日与平日各跑一次再定稿配置

7. 与「静态调优」的分工

手段	何时用
AQE coalesce / skewJoin	数据量波动大、轻中度倾斜、想少改代码
显式 repartition / pool	要稳定文件数、或 AQE 关不了的老集群
业务打散 / 两阶段	单键极端倾斜、AQE 拆分仍长尾
Broadcast / 桶表	维表足够小或 Join 键稳定且频繁

上线检查清单再补两条：配置中心是否覆盖 spark-defaults；AB 任务是否同版本 Spark（3.0 与 3.4 的 AQE 行为有差）。

常见坑（至少牢记这些）

1. 开了 AQE 却仍死守巨大静态 partitions
   初始分区过大时，合并有成本下限。initialPartitionNum / 上游分区要一起看。

2. 指望 AQE 解决极端热点
   单 key 占 50% 数据时仍要业务侧打散。AQE 是辅助不是银弹。

3. Broadcast 阈值盲目调大
   小表变大后 Driver/Executor OOM。用 UI 看 Broadcast 大小，设 autoBroadcastJoinThreshold 保守值。

4. 忽略 Spill 只看「能跑完」
   长期 Spill 等于隐式变慢与磁盘打满。先减分区膨胀、再加内存。

5. 本地调通、集群关 AQE
   平台默认 adaptive.enabled=false。提交参数与集群默认以 UI AdaptiveSparkPlan 为准。

如何验收（可量化）

• [ ] SQL Plan 含 AdaptiveSparkPlan；关键开关为 true
• [ ] 平日与活动日：总耗时、最长 Stage、Spill 有前后对比
• [ ] 倾斜 Scene：任务耗时比值下降或 skewJoin 拆分可见
• [ ] 写出文件数在目标范围（与小文件阈值一致）
• [ ] 结果 checksum（行数/金额）与调前一致

-- 结果验收
SELECT COUNT(*) AS cnt, SUM(cost_amount) AS cost
FROM dws_ad_channel_play_1d
WHERE dt = '${biz_date}';

# 从 Spark history / 日志摘指标（示意）
# stage_id, duration, shuffle_read_bytes, spill_bytes, num_tasks
# 对比 AQE 开关前后同一 stage 名称

指标	调前示例	调后目标
作业 P95	45min	≤20min（活动日）
最长 Stage 任务耗时比 Pmax/P50	>8	❤️
Spill	经常 >0	接近 0 或显著下降
输出文件数 / 分区	数千	数十～两百

相关面试题

占位：与题库小程序对齐后补链接（建议关联：AQE 原理、Shuffle 分区、倾斜处理）。

---

改写说明：结构参考 Wood 原创 Spark 调优实践，已改写成可执行清单；未包含未公开集群配置名。