关于spark shuffle过程的理解

　

shuffle

shuffle过程：由ShuffleManager负责，计算引擎HashShuffleManager（Spark 1.2）—>SortShuffleManager

spark根据shuffle类算子进行stage的划分，当执行某个shuffle类算子（reduceByKey、join）时，算子之前的代码被划分为一个stage，之后的代码被划分为下一个stage。当前stage开始执行时，它的每个task会从上一个stage的task坐在的节点通过网络拉取所需的数据

HashShuffleManager运行原理

未经优化的HashShuffleManager

shuffle write：

1. stage结束之后，每个task处理的数据按key进行“分类”
2. 数据先写入内存缓冲区
3. 缓冲区满，溢出到磁盘文件
4. 最终，相同key被写入同一个磁盘文件

创建的磁盘文件数量 = 当前stagetask数量 * 下一个stage的task数量

shuffle read：

1. 从上游stage的所有task节点上拉取属于自己的磁盘文件
2. 每个read task会有自己的buffer缓冲，每次只能拉取与buffer缓冲相同大小的数据，然后聚合，聚合完一批后拉取下一批
3. 该拉取过程，边拉取边聚合

优化后的HashShuffleManager

spark.shuffle.consolidateFiles设置为true可开启优化机制（consolidate机制）

• 一个Executor上有多少个core，就可并行执行多少个task
• 第一批并行执行的每个task会创建shuffleFileGroup，数据写入对应磁盘文件中
• 第一批task执行完后，下一批task复用已有的shuffleFileGroup

磁盘文件数量 = core数量 * 下一个stage的task数量

SortShuffleManager运行原理

普通运行机制

1. 数据先写入内存数据结构（reduceByKey—Map）
2. 每写一条数据，判断是否达到某个临界值，达到则根据key对数据排序，再溢出到磁盘
3. 写入磁盘通过java的Bufferedoutputstream实现，先缓存到内存，满后一次写入磁盘文件
4. 合并所有临时磁盘文件（merge），归并排序，依次写入同一个磁盘文件
5. 单独写一份索引文件，标识下游各个task的数据在文件中的start and end

磁盘文件数量 = 上游stage的task数量

bypass运行机制

当reducer的task数量  < spark.sort.bypassMergethreshold（200），shuffle write过程不排序，按未经优化的HashShuffleManager的方式write，最后将所有临时磁盘文件合并成一个，并创建索引文件

相关阅读

spark2.1.0自定义累加器AccumulatorV2的使用

类 类继承AccumulatorV2 class MyAccumulatorV2 extends AccumulatorV2[String, String] 覆写抽象方法： /** * @author lcjasa

JAVA的Collections类中shuffle的用法

就是随机打乱原来的顺序，和洗牌一样。如：// ShuffleTest.java import java.util.*; public class ShuffleTest { public stat

第9章Spark 2.1.0新一代Tungsten优化引擎彻底解析

第9章  Spark 2.1.0新一代Tungsten优化引擎彻底解析  1.1      概述Spark作为一个一体化多元化的大数据处理通用平台，性能

关于Collections.shuffle()方法的学习

Java.util.Collections类下有一个静态的shuffle()方法,如下：1）static void shuffle(List<?> list)  使用默认随机源对列表进行置换

关于Collections.shuffle()方法

Java.util.Collections类下有一个静态的shuffle()方法,如下： 1）static void shuffle(List<?> list)  使用默认随机源对列表进行置