Adaptive Execution 让 Spark SQL 更智能更高效

为了解决这个问题，Spark 新增接口，一次 Shuffle Read 可以读多个 Partition 的数据。如下图所示，Task 1 通过一轮请求即可同时读取 Task 0 内 Partition 0、1 和 2 的数据，减少了网络请求数量。同时 Mapper 0 一次性读取并返回三个 Partition 的数据，相当于顺序 IO，从而提升了性能。

由于 Adaptive Execution 的自动设置 Reducer 是由 ExchangeCoordinator 根据 Shuffle Write 统计信息决定的，因此即使在同一个 Job 中不同 Shuffle 的 Reducer 个数都可以不一样，从而使得每次 Shuffle 都尽可能最优。

上文 原有 Shuffle 的问题 一节中的例子，在启用 Adaptive Execution 后，三次 Shuffle 的 Reducer 个数从原来的全部为 3 变为 2、4、3。

2.4 使用与优化方法

可通过 spark.sql.adaptive.enabled=true 启用 Adaptive Execution 从而启用自动设置 Shuffle Reducer 这一特性

通过 spark.sql.adaptive.shuffle.targetPostShuffleInputSize 可设置每个 Reducer 读取的目标数据量，其单位是字节，默认值为 64 MB。上文例子中，如果将该值设置为 50 MB，最终效果仍然如上文所示，而不会将 Partition 0 的 60MB 拆分。具体原因上文已说明

3 动态调整执行计划

3.1 固定执行计划的不足

在不开启 Adaptive Execution 之前，执行计划一旦确定，即使发现后续执行计划可以优化，也不可更改。如下图所示，SortMergJoin 的 Shuffle Write 结束后，发现 Join 一方的 Shuffle 输出只有 46.9KB，仍然继续执行 SortMergeJoin

此时完全可将 SortMergeJoin 变更为 BroadcastJoin 从而提高整体执行效率。

3.2 SortMergeJoin 原理

SortMergeJoin 是常用的分布式 Join 方式，它几乎可使用于所有需要 Join 的场景。但有些场景下，它的性能并不是最好的。

SortMergeJoin 的原理如下图所示

• 将 Join 双方以 Join Key 为 Key 按照 HashPartitioner 分区，且保证分区数一致
• Stage 0 与 Stage 1 的所有 Task 在 Shuffle Write 时，都将数据分为 5 个 Partition，并且每个 Partition 内按 Join Key 排序
• Stage 2 启动 5 个 Task 分别去 Stage 0 与 Stage 1 中所有包含 Partition 分区数据的 Task 中取对应 Partition 的数据。(如果某个 Mapper 不包含该 Partition 的数据，则 Redcuer 无须向其发起读取请求)。
• Stage 2 的 Task 2 分别从 Stage 0 的 Task 0、1、2 中读取 Partition 2 的数据，并且通过 MergeSort 对其进行排序
• Stage 2 的 Task 2 分别从 Stage 1 的 Task 0、1 中读取 Partition 2 的数据，且通过 MergeSort 对其进行排序
• Stage 2 的 Task 2 在上述两步 MergeSort 的同时，使用 SortMergeJoin 对二者进行 Join

3.3 BroadcastJoin 原理

（编辑：惠州站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!