Adaptive Execution 让 Spark SQL 更智能更高效

本文转发自技术世界，原文链接　http://www.jasongj.com/spark/adaptive_execution/

1 背景

前面《Spark SQL / Catalyst 内部原理 与 RBO》与《Spark SQL 性能优化再进一步 CBO 基于代价的优化》介绍的优化，从查询本身与目标数据的特点的角度尽可能保证了最终生成的执行计划的高效性。但是

执行计划一旦生成，便不可更改，即使执行过程中发现后续执行计划可以进一步优化，也只能按原计划执行

CBO 基于统计信息生成最优执行计划，需要提前生成统计信息，成本较大，且不适合数据更新频繁的场景

CBO 基于基础表的统计信息与操作对数据的影响推测中间结果的信息，只是估算，不够精确

本文介绍的 Adaptive Execution 将可以根据执行过程中的中间数据优化后续执行，从而提高整体执行效率。核心在于两点

• 执行计划可动态调整
• 调整的依据是中间结果的精确统计信息

2 动态设置 Shuffle Partition

2.1 Spark Shuffle 原理

Spark Shuffle 一般用于将上游 Stage 中的数据按 Key 分区，保证来自不同 Mapper (表示上游 Stage 的 Task)的相同的 Key 进入相同的 Reducer (表示下游 Stage 的 Task)。一般用于 group by 或者 Join 操作。

如上图所示，该 Shuffle 总共有 2 个 Mapper 与 5 个 Reducer。每个 Mapper 会按相同的规则(由 Partitioner 定义)将自己的数据分为五份。每个 Reducer 从这两个 Mapper 中拉取属于自己的那一份数据。

2.2 原有 Shuffle 的问题

使用 Spark SQL 时，可通过 spark.sql.shuffle.partitions 指定 Shuffle 时 Partition 个数，也即 Reducer 个数

该参数决定了一个 Spark SQL Job 中包含的所有 Shuffle 的 Partition 个数。如下图所示，当该参数值为 3 时，所有 Shuffle 中 Reducer 个数都为 3

这种方法有如下问题

• Partition 个数不宜设置过大
• Reducer(代指 Spark Shuffle 过程中执行 Shuffle Read 的 Task) 个数过多，每个 Reducer 处理的数据量过小。大量小 Task 造成不必要的 Task 调度开销与可能的资源调度开销(如果开启了 Dynamic Allocation)
• Reducer 个数过大，如果 Reducer 直接写 HDFS 会生成大量小文件，从而造成大量 addBlock RPC，Name node 可能成为瓶颈，并影响其它使用 HDFS 的应用
• 过多 Reducer 写小文件，会造成后面读取这些小文件时产生大量 getBlock RPC，对 Name node 产生冲击
• Partition 个数不宜设置过小
• 每个 Reducer 处理的数据量太大，Spill 到磁盘开销增大
• Reducer GC 时间增长
• Reducer 如果写 HDFS，每个 Reducer 写入数据量较大，无法充分发挥并行处理优势
• 很难保证所有 Shuffle 都最优
• 不同的 Shuffle 对应的数据量不一样，因此最优的 Partition 个数也不一样。使用统一的 Partition 个数很难保证所有 Shuffle 都最优
• 定时任务不同时段数据量不一样，相同的 Partition 数设置无法保证所有时间段执行时都最优

2.3 自动设置 Shuffle Partition 原理

如 Spark Shuffle 原理 一节图中所示，Stage 1 的 5 个 Partition 数据量分别为 60MB，40MB，1MB，2MB，50MB。其中 1MB 与 2MB 的 Partition 明显过小(实际场景中，部分小 Partition 只有几十 KB 及至几十字节)

开启 Adaptive Execution 后

• Spark 在 Stage 0 的 Shuffle Write 结束后，根据各 Mapper 输出，统计得到各 Partition 的数据量，即 60MB，40MB，1MB，2MB，50MB
• 通过 ExchangeCoordinator 计算出合适的 post-shuffle Partition 个数(即 Reducer)个数(本例中 Reducer 个数设置为 3)
• 启动相应个数的 Reducer 任务
• 每个 Reducer 读取一个或多个 Shuffle Write Partition 数据(如下图所示，Reducer 0 读取 Partition 0，Reducer 1 读取 Partition 1、2、3，Reducer 2 读取 Partition 4)

三个 Reducer 这样分配是因为

• targetPostShuffleInputSize 默认为 64MB，每个 Reducer 读取数据量不超过 64MB
• 如果 Partition 0 与 Partition 2 结合，Partition 1 与 Partition 3 结合，虽然也都不超过 64 MB。但读完 Partition 0 再读 Partition 2，对于同一个 Mapper 而言，如果每个 Partition 数据比较少，跳着读多个 Partition 相当于随机读，在 HDD 上性能不高
• 目前的做法是只结合相临的 Partition，从而保证顺序读，提高磁盘 IO 性能
• 该方案只会合并多个小的 Partition，不会将大的 Partition 拆分，因为拆分过程需要引入一轮新的 Shuffle
• 基于上面的原因，默认 Partition 个数(本例中为 5)可以大一点，然后由 ExchangeCoordinator 合并。如果设置的 Partition 个数太小，Adaptive Execution 在此场景下无法发挥作用

由上图可见，Reducer 1 从每个 Mapper 读取 Partition 1、2、3 都有三根线，是因为原来的 Shuffle 设计中，每个 Reducer 每次通过 Fetch 请求从一个特定 Mapper 读数据时，只能读一个 Partition 的数据。也即在上图中，Reducer 1 读取 Mapper 0 的数据，需要 3 轮 Fetch 请求。对于 Mapper 而言，需要读三次磁盘，相当于随机 IO。

（编辑：淮北站长网）

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