|
Shuffle Write 结束后，可从每个 ShuffleMapTask 的 MapStatus 中统计得到按原计划执行时 Stage 2 各 Partition 的数据量以及 Stage 2 需要读取的总数据量。(一般来说，Partition 是 RDD 的属性而非 Stage 的属性，本文为了方便，不区分 Stage 与 RDD。可以简单认为一个 Stage 只有一个 RDD，此时 Stage 与 RDD 在本文讨论范围内等价)
如果其中一个 Stage 的数据量较小，适合使用 BroadcastJoin，无须继续执行 Stage 2 的 Shuffle Read。相反，可利用 Stage 0 与 Stage 1 的数据进行 BroadcastJoin，如下图所示
Spark SQL Auto BroadcastJoin
具体做法是
- 将 Stage 1 全部 Shuffle Write 结果广播出去
- 启动 Stage 2，Partition 个数与 Stage 0 一样，都为 3
- 每个 Stage 2 每个 Task 读取 Stage 0 每个 Task 的 Shuffle Write 数据，同时与广播得到的 Stage 1 的全量数据进行 Join
注：广播数据存于每个 Executor 中，其上所有 Task 共享，无须为每个 Task 广播一份数据。上图中，为了更清晰展示为什么能够直接 Join 而将 Stage 2 每个 Task 方框内都放置了一份 Stage 1 的全量数据
虽然 Shuffle Write 已完成，将后续的 SortMergeJoin 改为 Broadcast 仍然能提升执行效率
- SortMergeJoin 需要在 Shuffle Read 时对来自 Stage 0 与 Stage 1 的数据进行 Merge Sort，并且可能需要 Spill 到磁盘，开销较大
- SortMergeJoin 时，Stage 2 的所有 Task 需要取 Stage 0 与 Stage 1 的所有 Task 的输出数据(如果有它要的数据 )，会造成大量的网络连接。且当 Stage 2 的 Task 较多时，会造成大量的磁盘随机读操作，效率不高，且影响相同机器上其它 Job 的执行效率
- SortMergeJoin 时，Stage 2 每个 Task 需要从几乎所有 Stage 0 与 Stage 1 的 Task 取数据，无法很好利用 Locality
- Stage 2 改用 Broadcast，每个 Task 直接读取 Stage 0 的每个 Task 的数据(一对一)，可很好利用 Locality 特性。最好在 Stage 0 使用的 Executor 上直接启动 Stage 2 的 Task。如果 Stage 0 的 Shuffle Write 数据并未 Spill 而是在内存中，则 Stage 2 的 Task 可直接读取内存中的数据，效率非常高。如果有 Spill，那可直接从本地文件中读取数据，且是顺序读取，效率远比通过网络随机读数据效率高
3.5 使用与优化方法
该特性的使用方式如下
- 当 spark.sql.adaptive.enabled 与 spark.sql.adaptive.join.enabled 都设置为 true 时，开启 Adaptive Execution 的动态调整 Join 功能
- spark.sql.adaptiveBroadcastJoinThreshold 设置了 SortMergeJoin 转 BroadcastJoin 的阈值。如果不设置该参数，该阈值与 spark.sql.autoBroadcastJoinThreshold 的值相等
- 除了本文所述 SortMergeJoin 转 BroadcastJoin，Adaptive Execution 还可提供其它 Join 优化策略。部分优化策略可能会需要增加 Shuffle。spark.sql.adaptive.allowAdditionalShuffle 参数决定了是否允许为了优化 Join 而增加 Shuffle。其默认值为 false
4 自动处理数据倾斜
4.1 解决数据倾斜典型方案
《Spark性能优化之道——解决Spark数据倾斜(Data Skew)的N种姿势》一文讲述了数据倾斜的危害，产生原因，以及典型解决方法
- 保证文件可 Split 从而避免读 HDFS 时数据倾斜
- 保证 Kafka 各 Partition 数据均衡从而避免读 Kafka 引起的数据倾斜
- 调整并行度或自定义 Partitioner 从而分散分配给同一 Task 的大量不同 Key
- 使用 BroadcastJoin 代替 ReduceJoin 消除 Shuffle 从而避免 Shuffle 引起的数据倾斜
- 对倾斜 Key 使用随机前缀或后缀从而分散大量倾斜 Key，同时将参与 Join 的小表扩容，从而保证 Join 结果的正确性
4.2 自动解决数据倾斜
目前 Adaptive Execution 可解决 Join 时数据倾斜问题。其思路可理解为将部分倾斜的 Partition (倾斜的判断标准为该 Partition 数据是所有 Partition Shuffle Write 中位数的 N 倍) 进行单独处理，类似于 BroadcastJoin，如下图所示
Spark SQL resolve joinm skew
在上图中，左右两边分别是参与 Join 的 Stage 0 与 Stage 1 (实际应该是两个 RDD 进行 Join，但如同上文所述，这里不区分 RDD 与 Stage)，中间是获取 Join 结果的 Stage 2
明显 Partition 0 的数据量较大，这里假设 Partition 0 符合“倾斜”的条件，其它 4 个 Partition 未倾斜
以 Partition 对应的 Task 2 为例，它需获取 Stage 0 的三个 Task 中所有属于 Partition 2 的数据，并使用 MergeSort 排序。同时获取 Stage 1 的两个 Task 中所有属于 Partition 2 的数据并使用 MergeSort 排序。然后对二者进行 SortMergeJoin
（编辑：核心网）
【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!
|
