Spark AQE实战：从7小时到1小时的自适应查询优化调优指南

凌晨三点，手机震动，又是那个熟悉的值班群。一张 Spark UI 截图被甩进来：某个核心 ETL 任务已经跑了六个多小时，shuffle 阶段还有一大半 task 处于 pending 状态，而下游依赖它的报表再过两个小时就要开始刷新。这不是第一次了，每次上游数据一波动、热点 key 一转移，这条管道就会变成大型翻车现场。我所在的数据平台组维护着几千个离线 Spark 任务，这种场景见得太多。

直到我们把 Apache Spark 的自适应查询优化（Adaptive Query Execution，AQE）真正吃透，并在它基础上做了不少工程化改进，其中一个最头疼的任务终于从 7 小时压缩到了 1 小时，提效 7 倍。这篇文章就是把这套实践从原理到落地、从踩坑到二次改进，完完整整拆开讲一遍。如果你也在被 Spark 任务长尾、shuffle 爆炸、join 策略选错搞到焦头烂额，这篇内容应该能帮你省下不少折腾时间。

1. 静态执行计划为什么撑不住：从一次凌晨报警说起

1.1 执行计划是“猜”出来的，而统计信息经常说谎

在解释 AQE 之前，我们先看清楚传统 Spark 优化器是怎么工作的。Spark SQL 在生成执行计划时，会从元数据里读取表的统计信息，比如行数、总大小、列分布等。基于这些信息，优化器决定使用哪种 join 策略、每个 shuffle 分区多大、要不要做广播等。问题在于，这些统计信息大多是 Hive Metastore 里存的，而它经常是几天前甚至几周前的旧数据，而且过滤条件会大幅改变实际参与计算的数据量。

举个例子：某张维表在元数据里显示有 2 亿行、大小 120MB，超过广播阈值，优化器于是选择了 sort merge join。但实际查询里加了WHERE status = 'active'的过滤条件，参与 join 的数据只有 3000 万行、30MB，完全可以广播。Spark 没办法提前知道这一点，于是白白做了一个全量 shuffle，把几百 GB 的数据在集群里倒腾了一圈。这种“猜错”在生产环境里极其常见，也是静态执行计划的天花板。

1.2 那个跑 7 小时的任务：慢点到底在哪

我们重点优化的任务是一个标准的宽表构建流程：三张大表加一张维表做 join，输入数据约 20TB，中间 shuffle 落盘数据约 5TB。优化前的执行计划里有三处明显问题：

第一，维表过滤后实际只有 30MB，但优化器按统计信息选了 sort merge join，导致维表侧多产生了一次全量 shuffle。这一项就让任务额外多算了至少 40 分钟。

第二，另一个事实表与维度表 join 时，热点事件把某个 key 的数据量推到极端倾斜。其中一个 reduce task 要处理 12GB 的对应分区数据，其他 task 只有几百 MB。整个 stage 都在等这个“长尾”任务，单个 task 跑了 42 分钟。

第三，默认的spark.sql.shuffle.partitions=200把 reduce 阶段硬生生分成固定 200 个分区，完全不考虑数据分布。有些分区写入的数据只有几 MB，有些分区高达几个 GB，不均衡的副作用被进一步放大。

最终，任务总耗时 7 小时，其中 60% 以上时间是在等待倾斜 task 完成。业务方天天催，我们只能靠人工调整参数、手动加hint、甚至改业务 SQL 去规避，效率极低。

1.3 为什么需要“运行时自适应”而不是继续调优化器

静态优化器的核心缺陷在于，它只能基于“预测”做决策，而预测依赖的统计信息在生产环境几乎永远不够准确。AQE 的思路完全不同：它在执行过程中把计划分成多个 QueryStage，每个 Stage 跑完后，利用真实的 shuffle 输出统计信息（如每个分区大小、行数、key 分布），再决定接下来怎么走。这就好比开车时根据实时路况调整路线，而不是只靠出发前规划的导航。

Spark 3.0 引入 AQE 之后，我们第一时间在部分任务上试跑，确实有效，但距离“无脑开启”还很远。真正让这个 7 小时任务变成 1 小时，是结合了 AQE 的三大核心机制，再加上我们自己做的一些针对性改进。

2. AQE 三大核心机制拆解：动态合并分区、动态切换 join 策略、倾斜 join 优化

2.1 动态合并 shuffle 分区：减少无谓的 task 开销

Spark 默认会把 shuffle 的 reduce 分区数设置为spark.sql.shuffle.partitions，很多厂商为了省事直接留在默认值 200。如果上游 stage 的数据量很小，200 个 task 里可能有 60% 都在处理几十 KB 的空分区，白白浪费调度和序列化开销；如果数据量很大，200 个分区又不够，每个 task 要啃几 GB 的硬骨头。

AQE 的coalescePartitions机制会在 shuffle 写完之后，根据每个分区实际大小，把多个小分区合并成一个更大的分区。合并的目标是尽量接近spark.sql.adaptive.advisoryPartitionSizeInBytes（默认 64MB）。比如原来 200 个分区里有 180 个只有 5MB，AQE 会把这 180 个合并成约 14 个 64MB 的分区，reduce 阶段的总 task 数从 200 降到 20 左右。这不仅能大幅减少空转 task，还能减少最终输出的小文件数量。

这里有个容易被坑的点：coalescePartitions只会做“合并”，不会在初始分区数不足时帮你“增加”。如果初始分区数设得太低（比如还是 200），而实际 shuffle 数据有 5TB，每个分区平均 25GB，AQE 没能力把分区拆开。我们在实践中会根据任务历史 shuffle 数据量，把spark.sql.adaptive.coalescePartitions.initialPartitionNum设置成可能需要的分区数的 1.5 到 2 倍，给 AQE 留出合并空间。

2.2 动态切换 join 策略：把 sort merge join 悄悄改成 broadcast join

这是 AQE 给我们带来收益最大的一项。当 join 发生在两个 ShuffleQueryStage 之后，AQE 会根据每个 Stage 执行完的真实数据大小判断：如果小表侧的 shuffle 输出比广播阈值还小，它会把执行计划中的SortMergeJoin动态替换成BroadcastHashJoin。

原本两侧都要 shuffle，换成广播之后，只有大表侧继续做 shuffle，小表侧的数据被 driver 收集后通过 broadcast 变量发给所有 executor。这一步直接省掉一整份 shuffle 的读写和排序开销。以我们的任务为例，维表侧原本有约 500GB 的 shuffle 数据要写盘、传输、读盘，切换后这部分全部消失，任务瞬间轻了一多半。

注意，AQE 这里使用的广播阈值是spark.sql.adaptive.autoBroadcastJoinThreshold，如果没有显式设置，默认沿用spark.sql.autoBroadcastJoinThreshold（通常为 10MB）。由于 AQE 是在真实执行后判断，你可以放心地把这个阈值调大一点（比如 80MB 或 100MB），因为它只会在“确实足够小”的时候才广播，不会像静态优化器那样被过期统计信息误导。

2.3 倾斜 join 优化：把长尾拆成多条小尾巴

数据倾斜是离线任务最大的敌人。一个 key 占了 40% 的数据，意味着 reducer 端有一个 task 要处理四倍于其他 task 的数据，整个 stage 的时间被它拉满。AQE 的skewJoin专门处理这种场景。

它会扫描 shuffle 后各分区的数据量，如果某个分区的大小明显超过中位数的spark.sql.adaptive.skewJoin.skewedPartitionFactor倍（默认 5），同时超过绝对阈值spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes（默认 256MB），就把这个大表侧的分区拆成多个子分区，同时对小表侧对应分区进行复制，让多个 task 并行完成这组 join。

举个例子，大表的某个 key 在 shuffle 后有一个分区 12GB，其他分区平均 500MB。AQE 会把 12GB 拆成 10 个 1.2GB 的子分区，小表参与 join 的那个分区复制 10 份，分别 join。原来的 1 个长尾 task 变成 10 个小 task，并行度直接提高一个数量级。我们的任务里，那个 42 分钟的长尾 task 就靠这一项降到了 6 分钟左右。

3. 从 7 小时到 1 小时：一个真实任务的 AQE 调优全程

3.1 任务画像与瓶颈定位

在动手开启 AQE 之前，我们先花了一下午把这个任务从头到尾梳理了一遍。核心指标如下：

通过 Spark UI 里的 stage 耗时排序，我们确认了两个最大的热点：一是维表 join 阶段的 sort merge join 产生了巨额 shuffle，二是某个倾斜 key 导致的长尾 task。这两个问题不解决，再怎么加资源也没有用，因为瓶颈在单 task 的串行处理上。

3.2 关键参数配置：我们最终推荐的一组数值

我们使用的 Spark 版本是 3.2.1，以下是一组经过多次验证、适合中型规模任务（shuffle 量在 1TB 到 10TB 之间）的配置：

spark.sql.adaptive.enabled=true \ spark.sql.adaptive.coalescePartitions.enabled=true \ spark.sql.adaptive.coalescePartitions.initialPartitionNum=4000 \ spark.sql.adaptive.coalescePartitions.minPartitionNum=1 \ spark.sql.adaptive.advisoryPartitionSizeInBytes=128m \ spark.sql.adaptive.skewJoin.enabled=true \ spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 \ spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256m \ spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin=0.3 \ spark.sql.adaptive.autoBroadcastJoinThreshold=80m \ spark.sql.adaptive.localShuffleReader.enabled=true

这里有几个选择要特别说明：

• initialPartitionNum=4000：根据历史数据，该任务 shuffle 写 5TB，若想每个 reduce task 处理 1GB 左右，4000 个分区是合理的起点。AQE 后续会在此基础上合并小分区。
• advisoryPartitionSizeInBytes=128m：目标分区大小设为 128MB，而不是默认 64MB。过小的目标会让 reduce 阶段产生大量 task，调度开销反而变大。对于大多数分析型任务，128MB 是更稳妥的平衡点。
• autoBroadcastJoinThreshold=80m：调大广播阈值，但这只在 AQE 确认表足够小时才生效，安全边际比较高。
• skewedPartitionFactor=5和skewedPartitionThresholdInBytes=256m：保持默认。如果任务有特别集中的热点，可以适当调低 factor 到 3，但太激进的拆分反而会因广播复制小表数据带来额外内存压力。

3.3 优化前后的 DAG 与执行时间对比

开启 AQE 并调整参数之后，任务的表现发生了肉眼可见的变化。Spark UI 中执行计划里出现了AdaptiveSparkPlan标记，原本两个 ShuffleQueryStage 的 sort merge join，其中一个 stage 直接被 broadcast 替代。DAG 里少了一个完整的大 shuffle 阶段，shuffle 读数据从 5TB 降到了 2.8TB左右。

倾斜 join 的拆分也在日志里能看到。大表侧那个 12GB 的分区被自动切成 10 个子分区，加上复制后的维表分区，join 阶段多了 10 个并行 task，最长 task 时长降到了 6 分钟，没有再出现单点瓶颈。

最终，这个任务的总耗时稳定在 1 小时零 5 分左右，少数数据波动较大的日子也基本控制在 1.5 小时以内。对比原来的 7 小时，提速约 7 倍。

3.4 提效 7 倍背后的收益账本

从资源账来看，优化前任务平均占用 400 个 executor，每个 executor 4 核 16GB，总耗时 7 小时；优化后同样资源下跑 1 小时，节省的 CPU 时约 400 × 4 × 6 = 9600 核心时。如果按集群混部成本估算，一个任务每天省下近千个核心时，一个月就是几万核心时。更关键的是，任务窗口从凌晨 3 点提前到晚上 11 点前，下游报表再也不用来不及刷新的问题，值班电话少了一大半。

当然，收益并不全是 AQE 的功劳，也来自我们对静态参数的调整和对任务本身的理解。但 AQE 提供了一个系统性的框架，让这些优化不再依赖人肉调优，而是可以沉淀成平台能力。

4. 原生 AQE 不够用：我们在实践中的二次改进

4.1 统计信息“过期”之痛：强行在 AQE 前做一次轻量数据探查

AQE 虽然能根据运行时状态动态调整，但它对“哪个表可能变小”的初始判断仍然依赖统计信息。在我们的场景里，很多 Hive 表的统计信息长期不更新，导致 AQE 的广播阈值判断经常犹豫——它只能等 shuffle 实际执行完才知道小表变小了，但此时小表侧已经白白 shuffle 过一次，损失已经发生。

我们的解决办法是：在生成执行计划后、真正执行前，加入一个轻量数据探查规则。该规则扫描所有参与 join 的叶子表，如果元数据统计的最近更新时间超过 24 小时，就用spark.sql.adaptive的采样机制快速计算表的数据大小和行数，产生一个“校正后的统计信息”注入到计划中。采样本身耗时基本在 10 秒以内，但能让 AQE 在第一个 Stage 之前就有更准确的预期。这个改进让更多小表在一开始就被定为广播表，而不是等到 shuffle 完事后才反过来切换。

4.2 多列倾斜键识别：把单 key 优化扩展成复合 key 优化

原生的OptimizeSkewedJoin只处理单个等值 join 键的倾斜分区。但我们的业务里大量存在多列等值 join，比如事实表通过(country_id, channel_id)两个字段关联维表。单列视角下，每个 country 或每个 channel 的数据分布可能看不出明显倾斜，但复合在一起后，某些组合的局部数据量可能是中位数的几十倍，直接导致对应 task 长尾。

我们对 AQE 的倾斜检测逻辑做了扩展：在 shuffle 阶段，除了记录分区大小，还会对复合 join key 做一次分组统计，计算出每个 key 组合的数据量分布。如果某个组合的数据量超过中位数 factor 倍，并且超过阈值，就把该组合对应的整个分区对标记为倾斜，然后模仿原生拆分逻辑，把大分区分成子分区，小分区复制多份，并行 join。

这里要注意一个工程细节：多列倾斜检测需要对 shuffle 写入的数据做一次额外的预聚合，会增加约 5% 的 CPU 开销。我们通过一个配置项来控制是否开启，只对已知存在多列倾斜风险的任务启用，避免所有任务都背上这个包袱。

4.3 把动态合并、动态切策略、倾斜优化打成一套循环

原生 AQE 的几项优化是分别触发、一次性应用的。实际执行中我们发现，如果先做了分区合并，会抹平分区大小差异，导致倾斜检测误判；如果先做了倾斜拆分，又会改变最终分区大小，让后续的合并策略无从下手。我们把这些逻辑改造成一个循环迭代流程：

• 第一步：基于当前 Stage 统计信息，尝试切换 join 策略（sort merge -> broadcast）。
• 第二步：基于调整后的 shuffle 分区大小，执行动态分区合并。
• 第三步：对合并后仍显倾斜的分区执行拆分。
• 第四步：如果第三步产生了新的足够小的分区，回到第一步重新考虑是否有更多 join 可以被广播。

整个过程最多迭代三轮，并且每一轮都会记录决策日志，方便回溯。这个改进有效解决了“广播切换后，原本属于 broadcast 侧的倾斜 key 无法被识别”的问题，也让各项优化之间的配合更丝滑。

4.4 与物化视图和调度系统的集成

我们还做了一层偏工程的集成：当任务命中平台上的物化视图（MV）时，AQE 的决策会参考 MV 的产出时间和数据新鲜度。比如某张中间表已经有物化视图，且数据量足够小，就在计划阶段优先将对应的表替换为 MV，再把 AQE 的广播阈值放宽。这部分改动不大，但对整个数仓链路的运行稳定性帮助很明显——很多底层调度依赖都因为任务窗口缩短而变得更加宽容。

5. 落地 AQE 必须知道的事：参数、兼容性与排错指南

5.1 核心参数速查表

给刚接触 AQE 的团队一张可以直接抄的参数表，结合我们的实践经验，推荐值如下（基于 Spark 3.2+）：

5.2 版本与生态兼容性

AQE 从 Spark 3.0 开始正式引入，3.1 修复了不少 bug，3.2 之后稳定性明显提升，3.3/3.4 又加入了更智能的分区合并策略。如果你还在 Spark 2.4 上，想用 AQE 的唯一办法就是升级。我们内部也维护着 Spark 3.2 和 3.3 两个分支，实测 3.3 对倾斜 join 的细节拆分更优，值得投入。

同时要注意和周边组件的配合：如果使用 Kyuubi 作为 SQL 网关，需要同步升级客户端的 Spark 版本；如果依赖外部 shuffle service，要确认它支持 3.x 的push-based shuffle；如果用了 RAPIDS 之类的 GPU 加速插件，有些算子会绕过 AQE 的优化路径，需要额外适配。我们在落地时踩过不少这类兼容性的坑，建议先拿 10% 的任务灰度，不要一刀切。

5.3 常见问题排查：AQE 没生效？优化后反而变慢？

我们总结了三个高频问题，附上排查思路。

• AQE 设置后 Spark UI 里却看不到 AdaptiveSparkPlan：检查spark.sql.adaptive.enabled是否在任务提交时被覆盖，例如某些团队统一配置中心里把它关掉。另外，AQE 只对“至少含有一个 ShuffleQueryStage”的计划生效，纯逻辑查询或全部是 broadcast 的计划不会触发。
• 倾斜 join 没有生效：确认 join 类型是等值 join，且 join key 只有一个？多列需要我们的改进版本。再检查skewedPartitionThresholdInBytes是否大于实际倾斜分区的字节数，以及是否因为分区合并已经让倾斜不明显了。打开spark.sql.adaptive.logLevel=TRACE可以在日志里看到倾斜检测的详细判定。
• 开启 AQE 后任务反而更慢了：多半是advisoryPartitionSizeInBytes设置太小，导致 reduce 任务数量爆炸、调度和物化开销超过收益。另一个常见原因是autoBroadcastJoinThreshold调太大，导致一个明明有几 GB 的表被广播，driver 内存打爆。记住，AQE 的广播判断依赖真实大小，但它不会主动识别“广播这个表会导致 driver 内存不足”，这部分需要你自己守住上限。

5.4 灰度发布与效果评估

我们的标准流程是：先在测试环境用完整的历史数据重放一遍，对比开启 AQE 前后的 task 数量、shuffle 读数量、总耗时、失败率、GC 时间。然后在生产环境选取 20% 的任务作为灰度组，跑一周，看平均耗时和成功率，再逐步扩大到 100%。

一个很重要的细节：不要只比对总耗时。要深入看 stage 级别的耗时分布和长尾 task 数量，因为这些才是 AQE 发挥作用的核心指标。有的任务总耗时看起来差不多，但长尾从 40 分钟降到 10 分钟，说明资源峰值明显下降，整体集群的稳定性反而提升了。

6. 一些更深的思考：AQE 之后还有什么

6.1 AQE 带来的执行计划不确定性如何治理

AQE 让执行计划变成了“动态的”，这本来是好事，但也给问题定位带来了新挑战。同一个 SQL 今天可能走 broadcast，明天可能因为数据量变化走 sort merge，出问题时不好判断是 SQL 本身的问题还是计划变化导致的问题。

我们为此加了一个“计划变更审计”模块：AQE 每次做出策略切换、分区合并、倾斜拆分时，都会把决策原因和前后统计信息写入任务本地日志，最终汇总到平台。排查问题时，可以直接在日志里看到“某个 partition 从 800MB 被切成 6 个子分区，原因是超过中位数 4.2 倍”。这种可观测性非常重要，否则动态优化就变成一个黑盒，出了问题无从下手。

6.2 从 AQE 到运行时自适应架构

AQE 只是运行时自适应的一个开始。我们在实验把类似的思想应用到资源分配上：根据当前任务每个 stage 的实际 shuffle 数据量，动态调整 executor 的并行度、甚至调整队列资源。这有点像 AQE 的“扩大版”。虽然实现起来复杂，但收益也很明显——一个任务如果提前发现自己要处理的数据量只有预期的一半，完全可以提前归还部分资源给其他任务。

我们还尝试过把 AQE 和物化视图推荐打通：当 AQE 发现某个 join 的维表频繁被广播但数据量又刚好在阈值边缘时，就提示平台“这张表值得物化成小表”，或者建议上游做一次 compaction 压缩分区数。这个方向还在验证中，但思路很明确——让执行引擎成为优化数据治理的传感器。

6.3 给正在折腾 Spark 优化的团队的几个建议

第一，不要试图让所有任务都从一个推荐配置中受益。AQE 的每个参数都跟数据规模、集群规模、资源管理方式强相关。我们的建议是定义几组配置模板：小任务模板（shuffle < 100GB）、中任务模板（100GB ~ 3TB）、大任务模板（> 3TB），分别设置不同的initialPartitionNum和advisoryPartitionSizeInBytes。

第二，一定要把 AQE 的决策日志收集起来。只有在日志里看到“为什么这样改”，你才能持续优化你的 SQL 和表设计，而不是把调参变成玄学。

第三，如果条件允许，直接基于 Spark 3.3+ 做二次开发比在 3.0/3.1 上修 bug 更划算。我们最初在 3.1 上改了不少源码，后来升级到 3.3 时发现很多改动官方已经做了更好版本，白费了一部分功夫。

最后再分享一个实用技巧：在开启 AQE 的任务里，尽量把spark.sql.shuffle.partitions调大而不是依赖默认 200，因为 AQE 的“合并”只能在初始值基础上做减法。我们通常设置为 2000 或 4000，然后让advisoryPartitionSizeInBytes来引导最终分区规模。只要保证 AQE 有足够的“操作空间”，它会比任何人工静态调优都能更快地找到那个平衡点。