凌晨三点手机震动又是那个熟悉的值班群。一张 Spark UI 截图被甩进来某个核心 ETL 任务已经跑了六个多小时shuffle 阶段还有一大半 task 处于 pending 状态而下游依赖它的报表再过两个小时就要开始刷新。这不是第一次了每次上游数据一波动、热点 key 一转移这条管道就会变成大型翻车现场。我所在的数据平台组维护着几千个离线 Spark 任务这种场景见得太多。直到我们把 Apache Spark 的自适应查询优化Adaptive Query ExecutionAQE真正吃透并在它基础上做了不少工程化改进其中一个最头疼的任务终于从 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.partitions200把 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 个只有 5MBAQE 会把这 180 个合并成约 14 个 64MB 的分区reduce 阶段的总 task 数从 200 降到 20 左右。这不仅能大幅减少空转 task还能减少最终输出的小文件数量。这里有个容易被坑的点coalescePartitions只会做“合并”不会在初始分区数不足时帮你“增加”。如果初始分区数设得太低比如还是 200而实际 shuffle 数据有 5TB每个分区平均 25GBAQE 没能力把分区拆开。我们在实践中会根据任务历史 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 之前我们先花了一下午把这个任务从头到尾梳理了一遍。核心指标如下指标优化前输入数据量约 20TB三张大表 维表shuffle 写入量约 5TBreduce 分区数固定 200默认值最大单个 task 耗时42 分钟总耗时7 小时 10 分输出小文件超过 5000 个大量低于 5MB通过 Spark UI 里的 stage 耗时排序我们确认了两个最大的热点一是维表 join 阶段的 sort merge join 产生了巨额 shuffle二是某个倾斜 key 导致的长尾 task。这两个问题不解决再怎么加资源也没有用因为瓶颈在单 task 的串行处理上。3.2 关键参数配置我们最终推荐的一组数值我们使用的 Spark 版本是 3.2.1以下是一组经过多次验证、适合中型规模任务shuffle 量在 1TB 到 10TB 之间的配置spark.sql.adaptive.enabledtrue \ spark.sql.adaptive.coalescePartitions.enabledtrue \ spark.sql.adaptive.coalescePartitions.initialPartitionNum4000 \ spark.sql.adaptive.coalescePartitions.minPartitionNum1 \ spark.sql.adaptive.advisoryPartitionSizeInBytes128m \ spark.sql.adaptive.skewJoin.enabledtrue \ spark.sql.adaptive.skewJoin.skewedPartitionFactor5 \ spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256m \ spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin0.3 \ spark.sql.adaptive.autoBroadcastJoinThreshold80m \ spark.sql.adaptive.localShuffleReader.enabledtrue这里有几个选择要特别说明initialPartitionNum4000根据历史数据该任务 shuffle 写 5TB若想每个 reduce task 处理 1GB 左右4000 个分区是合理的起点。AQE 后续会在此基础上合并小分区。advisoryPartitionSizeInBytes128m目标分区大小设为 128MB而不是默认 64MB。过小的目标会让 reduce 阶段产生大量 task调度开销反而变大。对于大多数分析型任务128MB 是更稳妥的平衡点。autoBroadcastJoinThreshold80m调大广播阈值但这只在 AQE 确认表足够小时才生效安全边际比较高。skewedPartitionFactor5和skewedPartitionThresholdInBytes256m保持默认。如果任务有特别集中的热点可以适当调低 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参数名含义推荐值说明spark.sql.adaptive.enabledAQE 总开关true不解释必须开spark.sql.adaptive.coalescePartitions.enabled动态合并分区开关true一般建议开启spark.sql.adaptive.coalescePartitions.initialPartitionNum动态调整时的初始分区数不设则使用 shuffle.partitions多任务场景按数据量评估spark.sql.adaptive.coalescePartitions.minPartitionNum合并后的最小分区数1让 AQE 自由决定spark.sql.adaptive.advisoryPartitionSizeInBytes合并目标分区大小128m过大并行度低过小 task 多spark.sql.adaptive.autoBroadcastJoinThresholdAQE 判断广播的动态阈值80m或100m比静态阈值大胆一点spark.sql.adaptive.skewJoin.enabled倾斜 join 优化开关true对长尾任务尤其有用spark.sql.adaptive.skewJoin.skewedPartitionFactor倾斜判定倍数5调小更敏感需小心误判spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes倾斜判定绝对阈值256m小于此值不处理spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin非空分区比例阈值0.3防止空分区太多导致广播失衡spark.sql.adaptive.localShuffleReader.enabled本地 shuffle 读优化true减少不必要的远程读5.2 版本与生态兼容性AQE 从 Spark 3.0 开始正式引入3.1 修复了不少 bug3.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.logLevelTRACE可以在日志里看到倾斜检测的详细判定。开启 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 有足够的“操作空间”它会比任何人工静态调优都能更快地找到那个平衡点。 
长河君 · 2026/10/19 10:09:47 · 阅读 1.2k Spark AQE实战：从7小时到1小时的自适应查询优化调优指南
凌晨三点手机震动又是那个熟悉的值班群。一张 Spark UI 截图被甩进来某个核心 ETL 任务已经跑了六个多小时shuffle 阶段还有一大半 task 处于 pending 状态而下游依赖它的报表再过两个小时就要开始刷新。这不是第一次了每次上游数据一波动、热点 key 一转移这条管道就会变成大型翻车现场。我所在的数据平台组维护着几千个离线 Spark 任务这种场景见得太多。直到我们把 Apache Spark 的自适应查询优化Adaptive Query ExecutionAQE真正吃透并在它基础上做了不少工程化改进其中一个最头疼的任务终于从 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.partitions200把 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 个只有 5MBAQE 会把这 180 个合并成约 14 个 64MB 的分区reduce 阶段的总 task 数从 200 降到 20 左右。这不仅能大幅减少空转 task还能减少最终输出的小文件数量。这里有个容易被坑的点coalescePartitions只会做“合并”不会在初始分区数不足时帮你“增加”。如果初始分区数设得太低比如还是 200而实际 shuffle 数据有 5TB每个分区平均 25GBAQE 没能力把分区拆开。我们在实践中会根据任务历史 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 之前我们先花了一下午把这个任务从头到尾梳理了一遍。核心指标如下指标优化前输入数据量约 20TB三张大表 维表shuffle 写入量约 5TBreduce 分区数固定 200默认值最大单个 task 耗时42 分钟总耗时7 小时 10 分输出小文件超过 5000 个大量低于 5MB通过 Spark UI 里的 stage 耗时排序我们确认了两个最大的热点一是维表 join 阶段的 sort merge join 产生了巨额 shuffle二是某个倾斜 key 导致的长尾 task。这两个问题不解决再怎么加资源也没有用因为瓶颈在单 task 的串行处理上。3.2 关键参数配置我们最终推荐的一组数值我们使用的 Spark 版本是 3.2.1以下是一组经过多次验证、适合中型规模任务shuffle 量在 1TB 到 10TB 之间的配置spark.sql.adaptive.enabledtrue \ spark.sql.adaptive.coalescePartitions.enabledtrue \ spark.sql.adaptive.coalescePartitions.initialPartitionNum4000 \ spark.sql.adaptive.coalescePartitions.minPartitionNum1 \ spark.sql.adaptive.advisoryPartitionSizeInBytes128m \ spark.sql.adaptive.skewJoin.enabledtrue \ spark.sql.adaptive.skewJoin.skewedPartitionFactor5 \ spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256m \ spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin0.3 \ spark.sql.adaptive.autoBroadcastJoinThreshold80m \ spark.sql.adaptive.localShuffleReader.enabledtrue这里有几个选择要特别说明initialPartitionNum4000根据历史数据该任务 shuffle 写 5TB若想每个 reduce task 处理 1GB 左右4000 个分区是合理的起点。AQE 后续会在此基础上合并小分区。advisoryPartitionSizeInBytes128m目标分区大小设为 128MB而不是默认 64MB。过小的目标会让 reduce 阶段产生大量 task调度开销反而变大。对于大多数分析型任务128MB 是更稳妥的平衡点。autoBroadcastJoinThreshold80m调大广播阈值但这只在 AQE 确认表足够小时才生效安全边际比较高。skewedPartitionFactor5和skewedPartitionThresholdInBytes256m保持默认。如果任务有特别集中的热点可以适当调低 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参数名含义推荐值说明spark.sql.adaptive.enabledAQE 总开关true不解释必须开spark.sql.adaptive.coalescePartitions.enabled动态合并分区开关true一般建议开启spark.sql.adaptive.coalescePartitions.initialPartitionNum动态调整时的初始分区数不设则使用 shuffle.partitions多任务场景按数据量评估spark.sql.adaptive.coalescePartitions.minPartitionNum合并后的最小分区数1让 AQE 自由决定spark.sql.adaptive.advisoryPartitionSizeInBytes合并目标分区大小128m过大并行度低过小 task 多spark.sql.adaptive.autoBroadcastJoinThresholdAQE 判断广播的动态阈值80m或100m比静态阈值大胆一点spark.sql.adaptive.skewJoin.enabled倾斜 join 优化开关true对长尾任务尤其有用spark.sql.adaptive.skewJoin.skewedPartitionFactor倾斜判定倍数5调小更敏感需小心误判spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes倾斜判定绝对阈值256m小于此值不处理spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin非空分区比例阈值0.3防止空分区太多导致广播失衡spark.sql.adaptive.localShuffleReader.enabled本地 shuffle 读优化true减少不必要的远程读5.2 版本与生态兼容性AQE 从 Spark 3.0 开始正式引入3.1 修复了不少 bug3.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.logLevelTRACE可以在日志里看到倾斜检测的详细判定。开启 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 有足够的“操作空间”它会比任何人工静态调优都能更快地找到那个平衡点。