自适应查询执行（AQE）是在查询执行期间发生的查询重新优化。
在查询运行前，优化器会根据数据大小的估计选择查询计划。 如果表统计数据缺失或过时，数据有偏差，或者在复杂查询的后续步骤中，这些估计可能有误。 建立在错误估算上的计划可能采用慢速连接策略，在 Shuffle 后新增过多小任务，或者制造几个过大的任务而阻塞整个阶段。 AQE通过在查询运行时根据数据的实际大小重新优化计划来解决这些问题。
每次混洗或广播交换结束时，Azure Databricks 都拥有迄今为止所生成数据的准确统计数据。 AQE利用这些统计数据选择更好的物理策略，选择洗牌后分区的最优数量和大小，或应用否则原本需要查询提示的优化，如倾斜连接处理。
默认情况下，AQE 处于启用状态。 你不需要更改查询内容才能受益。 在排查慢查询时，检查AQE是否更改了查询计划。 只有当默认设置不适合你的工作负载时才调整AQE设置。 参见 查询计划 与 配置。
能力
AQE有四个主要特点：
- 将排序合并联接动态更改为广播哈希联接。
- 在随机交换后将分区进行动态联合（将小分区合并为大小合理的分区）。 非常小的任务的 I/O 吞吐量更差，并且往往受到计划开销和任务设置开销的影响。 合并小型任务可节省资源并提高群集吞吐量。
- 动态处理排序合并联接和随机哈希联接中的倾斜，方法是将倾斜的任务拆分（如果需要，还要进行复制）为大小大致相等的任务。
- 动态检测并传播空关系，空关系是查询中不返回行的部分。
受支持的查询
AQE适用于满足以下两个条件的查询：
- 这个查询不是流式查询。
- 查询至少包含一个 Exchange 运算符（通常来自连接、聚合或窗口）、至少一个子查询，或两者兼有。
AQE会重新优化所有支持的查询，但结果可能仍然是与优化器在查询运行前生成的计划相同的计划。 要检查AQE是否更改了某个查询的计划，请参见 查询计划。
查询计划
AQE 应用的查询包含一个或多个 AdaptiveSparkPlan 节点，通常作为每个主查询或子查询的根节点。 每个 AdaptiveSparkPlan 节点都有一个 isFinalPlan 标志。 该标志在查询执行前和执行期间为 false，并在查询结束时变更为 true。
要查看查询计划，请使用 Spark UI、DataFrame.explain()或SQL EXPLAIN。 每个都展示了AQE如何改变计划的不同信息。
Spark用户界面
Spark UI 中的查询计划图显示最新的计划，并会随着查询运行而更新。 已经运行过且有指标的节点不会改变。 尚未运行的节点可能会随着AQE重新优化计划而发生变化。
这个示例展示了查询计划图：
DataFrame.explain()
DataFrame.explain() 在每个 AdaptiveSparkPlan 节点下显示两个计划：初始计划（AQE 应用任何优化之前）和当前计划。 当前计划会随着查询的运行而变化，并在查询结束时成为最终计划。
输出包括每个 Shuffle 和广播阶段的数据统计数据。 统计数据取决于该阶段是否已运行：
- 在阶段运行前及期间，
isRuntime正在进行false，统计数据都是编译时估计。 例如，Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false)。 - 阶段运行后，
isRuntime为true，这些统计信息是运行时测量值。 例如，Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true)。
这些截图显示了 DataFrame.explain() 在执行前、执行中和执行后的输出：
执行之前
在执行过程中
执行后
SQL EXPLAIN
SQL EXPLAIN 不会执行查询，因此 AQE 无法重新优化它。 输出只显示初始计划，可能与AQE运行的计划不同。
这个示例展示了 SQL EXPLAIN 输出：
在查询计划中识别已应用的优化
当AQE应用优化时，查询计划会发生变化。 要确定哪些优化生效，可以比较当前或最终计划与初始计划。 每次优化都会以特定方式改变计划：
合并策略更改。 当前或最终方案的物理连接节点与初始计划不同。
分区合并。 该计划包含一个具有
Coalesced属性的CustomShuffleReader节点。
斜连接处理。 该计划包含一个
SortMergeJoin节点，字段isSkew设为true。
空关系传播。
LocalTableScan一个关系字段为空的节点替换了部分或全部计划。
配置
这些Apache Spark配置属性控制AQE。 大多数工作负载不需要更改默认设置。
启用AQE并调整各个优化
利用这些属性来开关AQE，并调整自动优化 shuffle、广播连接切换和空关系传播：
| 资产 | 类型 | Default | 说明 |
|---|---|---|---|
spark.databricks.optimizer.adaptive.enabled |
Boolean |
true |
可开启或关闭自适应查询执行。 |
spark.sql.shuffle.partitions |
Integer |
200 |
用于联接或聚合时进行数据重排的默认分区数。 将其设置为 auto 以启用自动优化 Shuffle，该功能会根据查询计划和查询输入数据大小设置此数值。对于结构化流，你不能在同一检查点位置进行查询重启时更改这个值。 |
spark.databricks.adaptive.autoBroadcastJoinThreshold |
Byte String |
30MB |
在运行时将排序合并连接切换为广播哈希连接的大小阈值。 |
spark.databricks.adaptive.emptyRelationPropagation.enabled |
Boolean |
true |
可开启或关闭动态空关系传播。 |
动态联合分区
| 资产 | 类型 | Default | 说明 |
|---|---|---|---|
spark.sql.adaptive.coalescePartitions.enabled |
Boolean |
true |
它会打开或关闭分区合并。 |
spark.sql.adaptive.advisoryPartitionSizeInBytes |
Byte String |
64MB |
联合后的目标大小。 聚合后的分区大小接近但不超过此大小。 |
spark.sql.adaptive.coalescePartitions.minPartitionSize |
Byte String |
1MB |
合并后分区的最小大小。 聚合后的分区大小不小于此值。 |
spark.sql.adaptive.coalescePartitions.minPartitionNum |
Integer |
集群核心数的两倍 | 合并后的最小分区数。 不要设置这个属性，因为设置它会覆盖 spark.sql.adaptive.coalescePartitions.minPartitionSize。 |
动态处理倾斜联接
| 资产 | 类型 | Default | 说明 |
|---|---|---|---|
spark.sql.adaptive.skewJoin.enabled |
Boolean |
true |
打开或关闭倾斜连接处理。 |
spark.sql.adaptive.skewJoin.skewedPartitionFactor |
Integer |
5 |
分区大小超过中位数分区大小多少倍时，该分区会被视为偏斜。 |
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes |
Byte String |
256MB |
被视为偏斜的分区必须超过的最小大小。 |
AQE 仅在这两个条件都成立时将分区视为偏斜：(partition size > skewedPartitionFactor * median partition size) 和 (partition size > skewedPartitionThresholdInBytes)。
常见问题 (FAQ)
为什么 AQE 未广播小型联接表？
如果表格小于 spark.databricks.adaptive.autoBroadcastJoinThreshold，但 AQE 不广播它，请检查以下原因：
- 连接类型不支持广播。 有些连接类型无法广播一侧。 例如，AQE 不能广播
LEFT OUTER JOIN的左侧。 - 该表有许多空分区。 排序合并连接中的大多数任务都能快速完成，或者通过偏斜连接处理来优化连接过程。 如果非空分区的比例低于
spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin，AQE不会将排序合并连接改为广播哈希连接。
我还应该在启用AQE的情况下使用广播加入提示吗？
是的，如果你对查询内容了解得很清楚。 优化器在查询运行前规划的广播连接通常比AQE在运行时规划的广播连接表现更好。 AQE可能要等到对连接两侧都进行 shuffle 之后，才会切换到广播连接，这时它才会得知实际的表大小。
AQE尊重查询提示，就像静态优化一样，同时仍然应用提示不影响的动态优化。
我应该使用倾斜连接 Hint 还是 AQE 倾斜连接处理？
使用AQE倾斜连接处理。 它是自动的，通常比 skew join 提示表现更好。
为什么 AQE 没有重新排序我的联接？
AQE在运行时不会重新排序连接。 为了改善连接顺序，收集表统计数据，使基于成本的优化器能够选择更好的方案。 参见 基于成本的优化器。
为什么 AQE 未检测到我的数据倾斜？
AQE仅在以下条件均成立时将划分视为偏斜：
- 该分区大于
256MB（默认值为spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes）。 - 分区大小大于中位分区大小的
spark.sql.adaptive.skewJoin.skewedPartitionFactor倍（默认5）。
AQE还仅在基于洗牌的连接中处理倾斜，这些连接包括排序合并连接和洗牌哈希连接。 它从不优化广播连接中的数据倾斜。 对于基于洗牌的连接，连接类型决定了AQE可以优化哪一侧：
| 联接类型 | 左侧倾斜已优化 | 右侧进行偏斜优化 |
|---|---|---|
INNER |
是的 | 是的 |
CROSS |
是的 | 是的 |
LEFT OUTER |
是的 | 否 |
RIGHT OUTER |
否 | 是的 |
LEFT SEMI |
是的 | 否 |
LEFT ANTI |
是的 | 否 |
FULL OUTER |
否 | 否 |
遗产
Spark 1.6 引入了一个叫自适应查询执行的功能，但 Spark 3.0 中的 AQE 是不同的功能。 Spark 1.6 的自适应执行仅动态合并分区。 AQE 是一个基于运行时统计来规划和重规划查询的框架，支持本页上的所有优化。