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。

（编辑：湖南网）

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