大数据：Spark 3.0 AQE 自适应查询执行怎么用？动态合并分区与自动拆分倾斜 Join

目录

一、为什么需要 AQE：静态计划的三个结构性缺陷

要理解 AQE 的价值，得先回到 Spark 2.x 时代的执行模型。Catalyst 优化器在提交作业前完成逻辑计划到物理计划的转换，随后整个 DAG 按固定分区数、固定 Join 策略、固定广播阈值一路跑到底。问题在于，优化器在编译期掌握的信息是"统计信息"，而统计信息天然滞后、易失真。这就带来三个结构性缺陷。

1.1 分区数靠拍脑袋

Spark SQL 的 spark.sql.shuffle.partitions 默认 200（Spark 3.x 仍为 200）。这个数字对中小作业偏大、对大作业偏小。分区过大导致大量小任务与调度开销，分区过小导致单任务数据量暴涨、GC 压力陡增。更麻烦的是，同一个作业在不同数据量下最优分区数完全不同，静态配置无法自适应。

本文评述：分区数本质上是"并行度"与"单任务负载"的权衡参数，它应当由运行时实际数据量反推，而非由开发者凭经验预设。AQE 的动态合并分区正是把这一决策从"编译期"推迟到"运行时"，这是思路上的根本转变。

1.2 数据倾斜无法预判

Join 或 GroupBy 时，某个 key 的记录数可能是均值的成百上千倍。静态计划对此无能为力——它不知道哪个分区会变成"长尾"。结果是少数任务拖垮整个 Stage，出现"99% 任务完成、1% 任务跑几小时"的经典长尾。

1.3 Join 策略与广播阈值僵化

广播 Join 的阈值 spark.sql.autoBroadcastJoinThreshold 默认 10MB，基于表统计信息判断。但统计信息可能缺失（如 Hive 表未 analyze）、可能过期（数据日增），导致该广播的没广播、不该广播的硬广播，前者退化为 SortMergeJoin，后者引发 OOM。

笔者认为：这三个缺陷的共同根源是"编译期信息不完备"。AQE 的解法不是让统计信息更准，而是承认信息不完备，在运行时用真实 shuffle 统计去修正计划。这是一种"务实主义"的工程哲学，比追求完美统计更可靠。

二、AQE 的架构与触发时序：QueryStage 是怎么跑起来的

2.1 从"一个 DAG"到"多个 QueryStage"

AQE 的核心抽象是 QueryStage。开启 AQE 后，物理计划在 Exchange（Shuffle）边界被切分成多个 Stage，每个 Stage 执行完成后会产出运行时统计（分区大小、记录数、倾斜分布），优化器据此重新规划后续 Stage。整个流程可概括为：

逻辑计划 → 物理计划 → [Stage1 执行] → 收集统计 → 重优化 → [Stage2 执行] → ... → 最终结果

关键点在于：重优化发生在 Stage 之间，而非 Stage 内部。这意味着 AQE 的收益取决于 Exchange 边界的数量与位置——Exchange 越多，可重优化的机会越多，但 Stage 切分本身也有开销。

2.2 三个核心组件

源码入口在 org.apache.spark.sql.execution.adaptive 包。开启开关是 spark.sql.adaptive.enabled=true（Spark 3.2 起默认 true）。

2.3 触发时序图（文字版）

┌─────────────┐    ┌──────────────┐    ┌───────────────┐
│ 提交物理计划 │ → │ 切分 QueryStage│ → │ 执行 Stage 1  │
└─────────────┘    └──────────────┘    └───────┬───────┘
                                               ↓
┌─────────────┐    ┌──────────────┐    ┌───────────────┐
│ 执行 Stage N │ ← │ 重优化物理计划 │ ← │ 收集运行时统计 │
└─────────────┘    └──────────────┘    └───────────────┘

本文评述：AQE 的"分阶段重优化"本质上是一种"在线查询优化"（Online Query Optimization）思想，与数据库领域的 Eddies、Rio 等自适应执行框架一脉相承。区别在于 Spark 的重优化粒度是 Stage 级而非 tuple 级，开销可控但灵活性稍弱。这个取舍在分布式批处理场景下是合理的。

三、动态合并 Shuffle 分区：从 200 个空跑到 20 个满载

3.1 机制原理

动态合并分区（Coalesce Shuffle Partitions）在 Shuffle 写阶段完成后，读取各 map 任务产出的分区大小统计，把相邻的小分区合并成不超过 spark.sql.adaptive.advisoryPartitionSizeInBytes（默认 64MB）的目标大小。合并只发生在同一 Shuffle 的相邻分区之间，不跨 Shuffle、不改变数据分布语义。

相关参数：

3.2 一个可复现的对比实验（模拟数据）

以下为笔者在本地 8 核 32GB 环境、Spark 3.3.2 上，用 TPC-DS 的 store_sales 表（scale factor 10，约 2.88 亿行）做的模拟对比。数据为整合模拟，仅用于说明趋势，不代表生产绝对值。

本文评述：合并分区不是"越小越好"或"越大越好"。目标 64MB 是 Spark 社区基于大量作业经验给出的折中值，但并非普适。对 CPU 密集型算子（如复杂 UDF），偏小的分区能提升并行度；对 IO 密集型算子，偏大的分区能减少小文件与调度开销。工程上应结合算子特征做 A/B。

3.3 操作步骤

1. 确认 spark.sql.adaptive.enabled=true 与 coalescePartitions.enabled=true；
2. 把 spark.sql.shuffle.partitions 适当调大（如 1000），让 AQE 有"向下合并"的空间；
3. 观察 Spark UI 的 SQL 页签，确认 Exchange 节点显示 "coalesced to N partitions"；
4. 根据 Stage 耗时与任务大小分布，微调 advisoryPartitionSizeInBytes。

官方文档参见 Spark SQL Performance Tuning；社区教程可参考 Spark SQL Reference。

四、自动拆分倾斜 Join：把"大 key"切成"小片"

4.1 倾斜的判定标准

AQE 在 Shuffle 读阶段统计每个分区的数据量，若某分区大小超过中位数的 spark.sql.adaptive.skewJoin.skewedPartitionFactor（默认 5）倍，且绝对值超过 skewedPartitionThresholdInBytes（默认 256MB），则判定为倾斜分区。

判定后，AQE 把倾斜分区按目标大小（advisoryPartitionSizeInBytes）拆成多个子分区，与另一侧对应分区做"一对多"的 Join，从而把单任务负载摊薄。

4.2 拆分逻辑示意

倾斜前：
  Partition P0 (2GB, key=A)  ×  Partition Q0 (key=A)
倾斜后：
  P0-1 (256MB) × Q0
  P0-2 (256MB) × Q0
  P0-3 (256MB) × Q0
  ...
  （Q0 被复制到多个子任务，代价是少量网络/内存放大）

本文评述：倾斜拆分的本质是"用空间换时间"——把另一侧分区复制多份，换取并行度。因此它并非万能：若两侧都严重倾斜，复制代价可能抵消收益。工程上应先判断倾斜是"单侧"还是"双侧"，双侧倾斜需结合 salting 等手动手段。

4.3 参数与调优

五、动态切换 Join 策略与广播阈值

5.1 动态降级广播 Join

当优化器基于统计信息选择了 BroadcastHashJoin，但运行时发现 build 侧实际大小超过 spark.sql.adaptive.autoBroadcastJoinThreshold（默认与静态阈值一致，10MB）时，AQE 会把 BroadcastHashJoin 降级为 SortMergeJoin，避免 OOM。这是"动态切换 Join 策略"最典型的应用。

5.2 动态选择 Join 策略

Spark 3.2 起引入 spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled 与更完整的 Join 策略重选逻辑，可在运行时根据实际大小在 BroadcastHashJoin、SortMergeJoin、ShuffledHashJoin 之间切换。ShuffledHashJoin 在 build 侧能放进内存且远小于 probe 侧时，往往比 SortMergeJoin 更快。

笔者认为：动态 Join 策略切换的价值常被低估。很多团队只关注"合并分区"和"倾斜拆分"，却忽略了广播降级——而后者恰恰是生产环境 OOM 的高频诱因。建议把 autoBroadcastJoinThreshold 与 AQE 联动检查，作为上线前的必查项。

六、参数体系与调优路径：一份可执行的 checklist

6.1 核心参数全景

6.2 调优五步法

1. 先开总开关，观察 SQL 页签是否出现 "AdaptiveSparkPlan"；
2. 看 Exchange 节点，确认分区是否被合并、合并到多少；
3. 看倾斜标记，Spark UI 会标注 "skewed" 分区；
4. 看 Join 策略，确认是否有广播降级；
5. 做 A/B，每次只改一个参数，记录 Stage 耗时与资源用量。

七、实战案例：三类场景的落地方法

7.1 电商大促：订单与商品维表关联

场景：订单表日增数亿行，商品维表约 50 万行。静态计划常因维表统计信息过期而误判，导致该广播的没广播。开启 AQE 后，动态广播阈值让维表在运行时被正确识别为小表，Join 从 SortMergeJoin 切换为 BroadcastHashJoin，Shuffle 数据量大幅下降。模拟数据显示 Stage 耗时下降约 40%（模拟数据，仅示意）。

7.2 日志聚合：GroupBy 长尾

场景：按 IP 聚合访问日志，少数爬虫 IP 记录数极高。AQE 的倾斜 Join 对 GroupBy 同样生效（本质是 PartialAggregate 后的 Shuffle），可把长尾分区拆开。若倾斜来自"空 key"或"默认值"，建议先做数据清洗，再依赖 AQE。

7.3 湖仓场景：Delta/Iceberg 上的 AQE

湖仓格式的分区裁剪与 AQE 有协同效应：文件级统计让 AQE 更准，AQE 又反过来减少小文件带来的分区碎片。近三年多篇工作（如 Delta Lake 的优化器集成、Iceberg 的 scan planning）都在探索这一方向，详见第九节。

八、边界、坑位与反模式

8.1 常见坑位

• 坑一：把 shuffle.partitions 设得极小，AQE 无合并空间，反而失去收益；
• 坑二：倾斜阈值设得过低，导致大量正常分区被误判拆分，复制开销反噬；
• 坑三：UDF 中持有大对象，AQE 合并分区后单任务内存压力反而上升；
• 坑四：依赖 AQE 解决所有倾斜，忽视 salting、两阶段聚合等手动手段。

8.2 反模式

反模式：把 AQE 当成"性能开关"，开了就完事。实际上 AQE 只解决"运行时信息不完备"这一类问题，对算子本身低效、数据模型设计不合理、资源配比失衡等问题无能为力。

九、前沿研究与演进预判

近三年（2022—2025）围绕自适应执行的研究主要集中在三个方向：一是把重优化粒度从 Stage 级细化到算子级，代表性工作包括对 Eddies 思想的再工程化；二是把学习型优化器（Learned Optimizer）与运行时统计结合，用历史执行反馈修正基数估计；三是湖仓格式与 AQE 的协同，利用文件级统计提升重优化精度。国内方面，阿里、字节、腾讯等团队在 Spark 内核上的倾斜治理与动态分区合并有大量工程实践沉淀，相关议题常见于 Spark Summit 与 Data + AI Summit 的分享。

本文评述：笔者认为，AQE 的下一步大概率不是"更激进的重优化"，而是"更聪明的代价模型"。当前 AQE 的决策规则（如 64MB 目标、5 倍倾斜因子）仍是启发式，未来若能把运行时统计喂给轻量级代价模型，决策质量会显著提升。但也要警惕过度工程化——启发式规则的可解释性与稳定性，在生产环境往往比"最优"更重要。

十、结语：把 AQE 当成"运行时治理"而非"开关"

AQE 的真正价值，不在于它替你省了几个参数，而在于它把"分区数、倾斜、Join 策略"这些原本靠经验拍板的决策，变成了可观测、可验证、可迭代的运行时治理过程。用好 AQE 的关键，是建立"看 UI → 定问题 → 调参数 → 做 A/B"的闭环，而不是迷信某个默认值。愿本文的路径与清单，能帮你在下一个大作业里少踩几个坑。

主要参考文献