一、AQE 是什么从静态到动态的范式转变1.1 传统 Spark 的问题Spark 2.x 的查询执行计划是静态的——在作业提交时就确定了全部执行计划基于的是统计信息估算而非真实数据。问题在于统计信息经常不准场景统计信息的问题读 Parquet 文件没有统计信息只能猜经过 Filter 后不知道过滤掉了多少行多表 Join 后行数估算误差累积UDF 处理后统计信息完全丢失1.2 AQE 的核心思想AQE 在Shuffle 边界插入动态优化点。Shuffle 是天然的执行断点——上游 Stage 完成后下游 Stage 启动前AQE 可以拿到上游真实的输出数据量据此调整下游执行计划。1.3 开启 AQE-- 开启 AQESpark 3.5 默认已开启 SET spark.sql.adaptive.enabled true; SET spark.sql.adaptive.coalescePartitions.enabled true; SET spark.sql.adaptive.skewJoin.enabled true; SET spark.sql.adaptive.localShuffleReader.enabled true;AQE 有三大核心能力能力配置项作用分区合并coalescePartitions小分区合并减少 Task 数倾斜 Join 处理skewJoin自动拆分倾斜分区动态 Join 切换autoBroadcastJoinSortMergeJoin → BroadcastJoin下面通过 10 个案例逐个演示。二、案例 1小文件分区合并2.1 问题现象-- 读取 20000 个小 Parquet 文件 SELECT region, COUNT(*) FROM events WHERE date 2026-08-01 GROUP BY region执行后发现有 20000 个 Task每个只处理不到 1MB 数据。Task 调度开销远超实际计算时间。-- 没开 AQE 的情况 Number of partitions: 20000 Average partition size: 0.8 MB Task scheduling overhead: ~45 min (20000 tasks × 0.13s) Actual compute time: ~3 min Total time: 48 min2.2 AQE 方案-- 开启分区合并目标分区大小 64MB SET spark.sql.adaptive.coalescePartitions.enabled true; SET spark.sql.adaptive.advisoryPartitionSizeBytes 67108864; -- 64MB SET spark.sql.adaptive.coalescePartitions.minPartitionSize 33554432; -- 32MB SET spark.sql.adaptive.coalescePartitions.initialPartitionNum 20000;2.3 效果-- 开启 AQE 后 Number of partitions: 248 (20000 → 248合并 80 倍) Average partition size: 62 MB Task scheduling overhead: ~0.5 min Actual compute time: ~3 min Total time: 4 min (从 48min → 4min提速 12 倍)三、案例 2数据倾斜导致 Task 长尾3.1 问题现象日志分析场景按user_id分组统计但某个超级用户产生了 70% 的日志SELECT user_id, COUNT(*) as cnt, SUM(bytes) as total_bytes FROM access_logs WHERE date 2026-08-01 GROUP BY user_id -- 没开倾斜处理的执行情况 Stage 1: 200 tasks Task 0 (user_id10086): 处理 15GB 数据耗时 38min Task 1-199 (其他用户): 各处理 75MB耗时 1min -- Stage 完成时间 max(38min, 1min) 38min3.2 AQE 方案-- 开启倾斜 Join 处理 SET spark.sql.adaptive.skewJoin.enabled true; SET spark.sql.adaptive.skewJoin.skewedPartitionFactor 5; -- 倾斜阈值分区大小 中位数 × 5 SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 268435456; -- 256MB3.3 AQE 倾斜处理原理3.4 效果倾斜分区被拆分为 4 个子分区 4 tasks × 3.75GB, each ~10min Stage 完成时间: 10min (从 38min → 10min, 提速 3.8x)四、案例 3动态 Join 策略切换4.1 问题现象大表 Join 小表但小表的大小在运行时才确定SELECT a.*, b.region_name FROM fact_orders a JOIN dim_region b ON a.region_id b.region_id dim_region 表统计信息显示有 5000 万行Spark 选择了 SortMergeJoin。但实际上经过 a.region_id b.region_id 的 Join 后dim_region 只剩 200 行因为很多 region 没有订单。 -- 静态计划 SortMergeJoin: 3h 20min BuildHashSort: 大表排序 2h BuildHashSort: 小表排序 0.5h ShuffleHashJoin: 0.5h4.2 AQE 方案SET spark.sql.adaptive.autoBroadcastJoinThreshold 104857600; -- 100MB -- 当 AQE 发现小表实际数据 100MB 时自动从 SortMergeJoin 切换为 BroadcastJoin4.3 AQE 动态切换原理静态计划: SortMergeJoin ↓ Stage 1 执行完毕 ↓ AQE 发现 dim_region 实际输出只有 200 行 (~5KB) ↓ 远小于 autoBroadcastJoinThreshold (100MB) ↓ 动态切换为 BroadcastJoin 动态计划: BroadcastHashJoin → 小表 Broadcast 到所有 Executor → 省去大表 Shuffle 和排序4.4 效果动态切换为 BroadcastJoin Broadcast 小表: 2s Map 端 Join: 12min Total: 12min (从 3h20min → 12min, 提速 16.7x)五、案例 4动态分区裁剪Dynamic Partition Pruning5.1 问题现象SELECT * FROM fact_sales f JOIN dim_store d ON f.store_id d.store_id WHERE d.region 华东fact_sales是按store_id分区的分区表有 1000 个分区。如果没有动态分区裁剪Spark 会扫描全部分区再过滤。5.2 AQE 方案SET spark.sql.optimizer.dynamicPartitionPruning.enabled true; -- Spark 3.5 中默认开启5.3 原理无 DPP: 扫描 fact_sales 全部 1000 个分区 → Join dim_store → Filter region华东 扫描数据量: 2TB 有 DPP: 先扫描 dim_store where region华东 → 得到 store_id 列表 [S001,S005,...] → 将 store_id 列表作为动态过滤条件 → 只扫描 fact_sales 中对应的分区 扫描数据量: 50GB (只扫描了 25 个分区)5.4 效果扫描分区数: 1000 → 25 (减少 97.5%) 扫描数据量: 2TB → 50GB 作业时间: 45min → 4min (提速 11x)六、案例 5多级聚合的分区优化6.1 问题现象SELECT province, city, district, SUM(amount) as total FROM orders GROUP BY province, city, district三级分组聚合默认会产生两层 Aggregatepartial final中间经过 Shuffle。6.2 AQE 方案SET spark.sql.adaptive.coalescePartitions.enabled true; SET spark.sql.adaptive.advisoryPartitionSizeBytes 134217728; -- 128MBAQE 会在第一层 Shuffle 后根据实际数据量合并分区避免第二层 Aggregate 产生过多小 Task。6.3 效果无 AQE: partial agg → 5000 分区 Shuffle → final agg 5000 Task (大量空分组) 有 AQE: partial agg → 5000 分区 Shuffle → AQE 合并为 80 分区 → final agg 80 Task final agg 时间: 18min → 2min Total: 32min → 16min (提速 2x)七、案例 6-10更多生产场景7.1 案例 6Shuffle 后分区数过多-- 大表 Join 后的 Shuffle 分区过多 SET spark.sql.adaptive.coalescePartitions.enabled true; SET spark.sql.adaptive.advisoryPartitionSizeBytes 67108864; SET spark.sql.adaptive.coalescePartitions.parallelismFirst false; -- parallelismFirstfalse 让 Spark 优先按数据大小而非分区数决定合并指标无 AQE有 AQEShuffle 分区数2000156空分区数18500作业时间25min8min7.2 案例 7倾斜 Join 的拆分粒度调优-- 默认阈值可能不够灵活 SET spark.sql.adaptive.skewJoin.skewedPartitionFactor 3; -- 降低阈值更积极拆分 SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 134217728; -- 128MB -- 对于极端倾斜场景 SET spark.sql.adaptive.skewJoin.skewedPartitionFactor 2;效果极端倾斜单个分区 50GB场景下从 52min 降至 14min。7.3 案例 8自动调整 SparkSQL 的 Broadcast 阈值-- 静态阈值 SET spark.sql.autoBroadcastJoinThreshold 10485760; -- 10MB (静态) -- AQE 动态阈值 SET spark.sql.adaptive.autoBroadcastJoinThreshold 104857600; -- 100MB (动态) -- 关键区别: -- 静态阈值基于统计信息估算的小表大小 -- 动态阈值基于 Shuffle 后实际的小表大小效果Join 性能提升 3-5 倍针对统计信息不准的场景。7.4 案例 9本地 Shuffle Reader 优化SET spark.sql.adaptive.localShuffleReader.enabled true; -- 当 BroadcastJoin 被触发后上游 Shuffle 的数据不需要真正 Shuffle -- AQE 会用 LocalShuffleReader 直接读本地数据 无 LocalShuffleReader: Stage 1 → Shuffle Write → 网络传输 → Shuffle Read → BroadcastJoin 有 LocalShuffleReader: Stage 1 → 本地写入 → 本地读取 → BroadcastJoin (省去网络传输)效果减少网络 IO在 200 个 Executor 的集群上节省约 8min 的 Shuffle 时间。7.5 案例 10AQE 与其他优化配合-- AQE 动态分区裁剪 文件合并组合使用 SET spark.sql.adaptive.enabled true; SET spark.sql.adaptive.coalescePartitions.enabled true; SET spark.sql.adaptive.skewJoin.enabled true; SET spark.sql.adaptive.localShuffleReader.enabled true; SET spark.sql.optimizer.dynamicPartitionPruning.enabled true; SET spark.sql.files.maxPartitionBytes 268435456; -- 256MB 文件合并 SET spark.sql.adaptive.advisoryPartitionSizeBytes 67108864; -- 64MB Shuffle 合并效果一个复杂 ETL 作业5 个 Join 3 个聚合从 2.5h 降至 28min。八、AQE 调优参数速查表参数默认值推荐值说明spark.sql.adaptive.enabledtruetrueAQE 总开关coalescePartitions.enabledtruetrue分区合并开关advisoryPartitionSizeBytes64MB64-128MB目标分区大小coalescePartitions.minPartitionSize1MB32MB最小分区大小skewJoin.enabledtruetrue倾斜 Join 开关skewJoin.skewedPartitionFactor53-5倾斜判定倍数skewedPartitionThresholdInBytes256MB128-256MB倾斜判定阈值autoBroadcastJoinThreshold10MB100MB动态 Broadcast 阈值localShuffleReader.enabledtruetrue本地 Shuffle 读取九、AQE 的局限性与避坑指南9.1 AQE 不是万能的局限说明替代方案只在 Shuffle 边界生效没有 Shuffle 的作业无法优化手动repartition不优化非 SQL APIRDD API 不走 Catalyst尽量用 DataFrame API统计信息仍然重要AQE 依赖 Shuffle 后的真实数据维护好表统计倾斜拆分有上限拆分次数有限极端倾斜仍可能长尾数据预处理9.2 常见踩坑# 坑1: RDD 操作不走 AQE rdd sc.textFile(data.txt) # AQE 不生效 df spark.read.text(data.txt) # AQE 生效 # 坑2: cache 后 AQE 优化可能被缓存绕过 df spark.sql(SELECT ...).cache() # 第一次执行: AQE 生效 # 后续执行: 直接读缓存AQE 的动态优化不生效 # 建议: 性能优化阶段不要 cache上线后再 cache # 坑3: AQE 与 AQE 不可控的场景 # 某些复杂子查询可能不走 AQE # 用 EXPLAIN 确认执行计划 spark.sql(SET spark.sql.adaptive.enabledtrue) df spark.sql(SELECT ...) df.explain(modeadaptiveCost) # 查看 AQE 生效后的计划十、总结AQE 是 Spark 3.x 最重要的性能优化机制。10 个案例的共性结论场景AQE 能力平均提速小文件/分区过多分区合并5-12x数据倾斜倾斜分区拆分3-4xJoin 策略不当动态 Broadcast 切换3-16x分区表扫描动态分区裁剪5-11x多级聚合分区合并优化2x实践建议生产环境务必开启 AQE 全部功能参数从默认值开始针对特定作业通过 EXPLAIN 分析后微调。下一篇预告下一篇我们回到 AI 领域深入 vLLM 的 Continuous Batching连续批处理源码解析它如何在不中断生成过程的情况下动态插入新请求实现 5800 t/s 的极致吞吐量。往期回顾Kafka acks 机制性能实测acksall 在百万级吞吐下的延迟代价有多大Flink 状态后端选型RocksDB vs Heap 在百万级吞吐下的 5 倍性能差异Kafka 深度解剖 2消费者组再均衡 Rebalance 全流程觉得有帮助请点赞收藏。关注专栏「AI大模型大数据硬件编程」每周更新大数据性能调优的深度内容。