上一篇解决了 DataFrame 和 Dataset 的 API 演进。这一篇进入运行时优化——Adaptive Query Execution（AQE）。

传统查询优化器的一个根本问题是：优化发生在执行之前，依赖的统计信息可能不准确或完全缺失。AQE 把优化推迟到执行过程中——在每个 Shuffle 边界，收集实际的数据统计信息，然后用这些精确的统计重新优化后续 Stage 的执行计划。

本文只抓一个问题：AQE 的三个核心优化分别解决什么问题，以及运行时优化的反馈循环如何工作。

运行时反馈循环

AQE 的工作方式可以用一句话概括：执行一个 Stage → 收集 Shuffle 输出的统计信息 → 用统计信息重新优化下一个 Stage 的计划 → 执行下一个 Stage。

1
2
3
4
5
6
7
8
9
10
11
12
传统执行:
  编译期确定完整计划 → Stage 0 → Stage 1 → Stage 2
  (统计信息可能不准，计划一旦确定不可修改)

AQE 执行:
  编译期确定初始计划 → 执行 Stage 0
      → 收集 Stage 0 的 Shuffle 输出统计
      → 重新优化 Stage 1 的计划
      → 执行 Stage 1
      → 收集 Stage 1 的 Shuffle 输出统计
      → 重新优化 Stage 2 的计划
      → 执行 Stage 2

Shuffle 是天然的观测点：上游 Stage 的所有 Task 必须全部完成后，下游 Stage 才能开始。在这个等待窗口中，AQE 可以统计每个 Shuffle 分区的实际数据量，然后据此调整计划。

优化一：合并小分区

spark.sql.shuffle.partitions 的默认值是 200。对于小数据量的查询，200 个分区可能导致大量的微小 Task，每个 Task 只处理几 KB 数据，调度开销远超计算开销。

AQE 的 Coalescing Post-Shuffle Partitions 优化在 Shuffle 完成后检查每个分区的实际数据量，把相邻的小分区合并成更大的分区：

1
2
3
4
5
6
7
Shuffle 输出 (200 个分区，大部分很小):
  P0: 1KB  P1: 2KB  P2: 0KB  P3: 1KB ... P199: 3KB

AQE 合并后 (20 个分区，每个约 64MB):
  P0': [P0-P9 合并] 15KB
  P1': [P10-P19 合并] 20KB
  ...

合并的目标大小由 spark.sql.adaptive.advisoryPartitionSizeInBytes（默认 64MB）控制。合并后 Task 数量减少，每个 Task 处理更多数据，减少了调度开销。

这个优化让用户不再需要精确调整 spark.sql.shuffle.partitions——设一个较大的值（如 200），AQE 会自动把多余的小分区合并掉。

优化二：动态切换 Join 策略

编译期选择 Join 策略依赖表的大小统计。如果统计信息缺失或不准确（比如经过过滤后表变小了），可能选择了 SortMergeJoin 而实际上 BroadcastHashJoin 更优。

AQE 在 Shuffle 后知道了 Join 输入的实际大小。如果某一侧的数据量小于广播阈值，AQE 把 SortMergeJoin 动态切换为 BroadcastHashJoin：

1
2
3
4
5
6
7
8
9
10
11
12
13
编译期计划 (统计信息显示两表都很大):
  SortMergeJoin
  ├── Shuffle (table_a)   预估 500MB
  └── Shuffle (table_b)   预估 200MB

实际执行 (table_b 经过过滤后很小):
  Shuffle table_a 完成: 实际 500MB
  Shuffle table_b 完成: 实际 8MB  ← 小于广播阈值 10MB

AQE 改写后的计划:
  BroadcastHashJoin
  ├── Shuffle (table_a)
  └── Broadcast (table_b)  ← 不再 Shuffle，直接广播

BroadcastHashJoin 比 SortMergeJoin 快得多——它避免了大表的排序和归并，直接用小表构建 hash 表在大表端做探测。

优化三：处理倾斜 Join

数据倾斜是分布式 Join 的经典难题：某些 key 的数据量远超平均值，导致处理这些 key 的 Task 耗时远超其他 Task。

AQE 的 Skew Join Optimization 在 Shuffle 后检测倾斜分区（数据量超过中位数的 N 倍，默认 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB），然后把倾斜分区拆成多个子分区，每个子分区与另一侧的对应分区做 Join：

1
2
3
4
5
6
7
8
9
10
11
倾斜场景:
  table_a 分区: P0(10MB)  P1(10MB)  P2(500MB)  P3(10MB)
  table_b 分区: P0(5MB)   P1(5MB)   P2(5MB)    P3(5MB)
  
  P2 是倾斜分区——500MB vs 平均 10MB

AQE 拆分后:
  table_a: P0(10MB)  P1(10MB)  P2-a(167MB) P2-b(167MB) P2-c(166MB)  P3(10MB)
  table_b: P0(5MB)   P1(5MB)   P2(5MB)     P2(5MB)     P2(5MB)      P3(5MB)
                                  ↑           ↑           ↑
                           table_b 的 P2 被复制到每个子分区

拆分后，原来一个 Task 处理 500MB 数据，变成三个 Task 各处理约 167MB + 5MB。总数据量略有增加（table_b 的 P2 被复制了），但并行度提升，消除了长尾 Task。

实验：观察 AQE 效果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
val spark = SparkSession.builder.getOrCreate()
import spark.implicits._

// 构造倾斜数据
val skewed = (1 to 1000000).map(i => if (i % 100 == 0) (1, i) else (i, i))
val left = spark.createDataFrame(skewed).toDF("key", "val_a")
val right = spark.range(1, 1001).toDF("key", "val_b")

// AQE 开启（Spark 3.0+ 默认开启）
spark.conf.set("spark.sql.adaptive.enabled", "true")
val result1 = left.join(right, "key")
result1.explain(true)
result1.count()

// AQE 关闭
spark.conf.set("spark.sql.adaptive.enabled", "false")
val result2 = left.join(right, "key")
result2.count()

// 在 Spark UI 对比两次执行的:
// 1. Stage 数量（AQE 可能合并 Stage 或改变 Join 策略）
// 2. Task Duration 分布（AQE 的 skew 优化应该减少最长 Task 时间）
// 3. Shuffle 分区数（AQE 的合并优化应该减少实际分区数）

AQE 开启时，Spark UI 的 SQL 页面会显示 “AdaptiveSparkPlan” 节点，表示该查询使用了自适应优化。点击进入可以看到编译期计划和实际执行计划的差异。

模式提炼

AQE 体现了运行时自适应优化的通用模式：用执行过程中的实际数据替代编译期的预估数据来做决策。

这个模式在多个领域出现。JIT 编译器在运行时收集分支预测和热点方法信息，据此做内联和去虚化优化。TCP 的拥塞控制在传输过程中测量 RTT 和丢包率，动态调整窗口大小。自适应负载均衡器在请求过程中测量后端响应时间，动态调整流量分配。

核心思路一致：编译期/配置期的信息是静态的、可能不准确的；运行时的观测数据是动态的、精确的。在系统有能力在运行时做出调整的地方，用动态数据替代静态预估。

工程迁移表

Spark/AQE 概念 传统数据库对应 Flink 对应 通用系统对应
运行时重新优化 自适应查询处理（AQP） 无直接对应 JIT 编译的 profile-guided optimization
合并小分区 无（分区数静态确定） 无（并行度静态确定） 自动 auto-scaling
动态切换 Join 策略 运行时 Join 方法切换（如 Oracle AQP） 无 A/B 测试中的多臂老虎机
倾斜 Join 处理 倾斜处理 hint（手动） 无内置自动处理 负载均衡的热点分散
Shuffle 边界统计收集 执行期间的基数估算 Operator 的 watermark/metrics 监控系统的实时指标采集

常见误解

误解一：“AQE 可以修复任意的查询性能问题”。AQE 只在 Shuffle 边界做优化。如果性能瓶颈在 Scan 阶段（读取大量不需要的数据）或单个 Task 内部（复杂的 UDF 计算），AQE 帮不上忙。

误解二：“开启 AQE 后不再需要调整 spark.sql.shuffle.partitions”。AQE 的合并小分区确实减轻了对这个参数的敏感度，但过小的初始值（比如 1）仍然会导致 Shuffle 阶段只有一个 Task——AQE 只能合并，不能拆分已有的分区（拆分只发生在 Skew Join 场景）。

误解三：“AQE 没有开销”。AQE 在每个 Shuffle 边界需要收集统计信息并重新运行优化器，这带来少量的延迟（通常几十毫秒）。对于亚秒级的微小查询，这个延迟可能不可忽略。

练习

  1. 执行一条涉及两表 Join 的 SQL，在 Spark UI 的 SQL 页面查看 AdaptiveSparkPlan 节点。对比 “Initial Plan” 和 “Final Plan” 的差异。Join 策略是否发生了变化？

  2. 设置 spark.sql.shuffle.partitions=1000，执行一个小数据量的聚合查询（如 1000 行数据的 groupBy）。在 Spark UI 中观察 AQE 合并后的实际 Task 数量。

  3. 构造一个极度倾斜的数据集（90% 的数据 key 相同），做 Join 并对比 AQE 开启/关闭时最长 Task 和最短 Task 的耗时比。

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 已完成
06 存储体系：Block Manager、广播变量与累加器 已完成
07 容错机制：Lineage、Checkpoint 与推测执行 已完成
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树 已完成
09 DataFrame 与 Dataset：类型安全与性能的折中 已完成
10 Adaptive Query Execution：运行时改写执行计划 本篇
11 数据源 API 与谓词下推：让存储层帮忙过滤 下一篇
12 Structured Streaming：把流当成无界表
13 水位线与状态管理：流处理的时间语义
14 Spark on YARN / K8s：资源管理与部署模式
15 动态资源分配与作业调度：多租户集群的资源博弈
16 性能诊断：Spark UI、Event Log 与调优思路
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料