AQE 是什么
AQE（自适应查询执行）让 Spark 在运行时根据 shuffle 之后拿到的真实分区统计，回头调整执行计划：把大量小分区合并成合理大小、把 sort-merge join 改成广播 join、把异常大的分区拆成多个子任务处理。它解决的是「写 SQL 时估不准数据量」这个老问题——静态优化只能靠表统计信息猜，AQE 用实际数据修正。Spark 3.2 起 spark.sql.adaptive.enabled 默认为开启状态。
基本信息
| 总开关 | spark.sql.adaptive.enabled，Spark 3.2 起默认开启 |
|---|---|
| 三项主要能力 | 合并小分区、运行时切换 join 策略、拆分倾斜分区 |
| 目标分区大小 | spark.sql.adaptive.advisoryPartitionSizeInBytes，默认 64MB |
| 倾斜开关 | spark.sql.adaptive.skewJoin.enabled，默认开启 |
| 倾斜判据 | 分区大小超过中位数若干倍且超过绝对阈值 |
| 相关能力 | 动态分区裁剪，把 join 结果下推到扫描侧减少读取 |
| 局限 | 只作用于 shuffle 之后的阶段，扫描过多与极端倾斜仍需业务侧处理 |
典型使用场景
- 数据量波动大的作业：靠运行时合并小分区，省掉为每次数据量手调分区数
- 大表关联小表：运行阶段发现小侧足够小就改走广播，避免一次多余 shuffle
- 分区大小极度不均：把超大的倾斜分区拆成多个子任务，缩短长尾
- 多表 join 链路：动态分区裁剪把过滤下推到扫描阶段，减少读取的数据量
- 版本升级：升级到 3.2 及以上后，历史作业里为绕开问题加的手工参数可以逐步清理
AQE 配置示例
AQE 配置示例
spark.conf.set('spark.sql.adaptive.enabled', 'true') # Spark 3.2 起默认开启
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true') # 合并小分区
spark.conf.set('spark.sql.adaptive.advisoryPartitionSizeInBytes', '128m') # 合并后的目标分区大小
spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true') # 拆分倾斜分区
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionFactor', '5') # 超过中位数 5 倍视为倾斜
spark.conf.set('spark.sql.adaptive.localShuffleReader.enabled', 'true') # 小数据量时本地读 shuffle
df = spark.sql('SELECT store_id, sum(pay_amt) AS amt FROM dwd_trade_order_di GROUP BY store_id')
df.explain('formatted') # 查看 AdaptiveSparkPlan 节点确认 AQE 生效主要子开关
| 子开关 | 作用 | 默认值 |
|---|---|---|
| adaptive.coalescePartitions | 运行时合并小分区 | 开启 |
| adaptive.skewJoin | 拆分倾斜分区并处理小侧 | 开启 |
| adaptive.localShuffleReader | 小结果时本地读取 shuffle 数据 | 开启 |
| advisoryPartitionSizeInBytes | 合并后的目标分区大小 | 64MB |
开启 AQE 后，shuffle 分区数变成「初始分区数」，最终分区由运行时合并或拆分决定。初始值仍要设在合理量级，否则一开始就产生几十万个微小分区，合并本身也要付出代价。
AQE 只能优化 shuffle 之后的问题。扫描阶段读取过多、单个 key 占据大半数据的极端倾斜、以及上游数据质量问题，仍然会表现为任务超时或内存溢出。