关于spark shuffle过程的理解
shuffle过程:由ShuffleManager负责,计算引擎HashShuffleManager(Spark 1.2)—>SortShuffleManager
spark根据shuffle类算子进行stage的划分,当执行某个shuffle类算子(reduceByKey、join)时,算子之前的代码被划分为一个stage,之后的代码被划分为下一个stage。当前stage开始执行时,它的每个task会从上一个stage的task坐在的节点通过网络拉取所需的数据
HashShuffleManager运行原理
未经优化的HashShuffleManager
shuffle write:
- stage结束之后,每个task处理的数据按key进行“分类”
- 数据先写入内存缓冲区
- 缓冲区满,溢出到磁盘文件
- 最终,相同key被写入同一个磁盘文件
创建的磁盘文件数量 = 当前stagetask数量 * 下一个stage的task数量
shuffle read:
- 从上游stage的所有task节点上拉取属于自己的磁盘文件
- 每个read task会有自己的buffer缓冲,每次只能拉取与buffer缓冲相同大小的数据,然后聚合,聚合完一批后拉取下一批
- 该拉取过程,边拉取边聚合
优化后的HashShuffleManager
spark.shuffle.consolidateFiles设置为true可开启优化机制(consolidate机制)
- 一个Executor上有多少个core,就可并行执行多少个task
- 第一批并行执行的每个task会创建shuffleFileGroup,数据写入对应磁盘文件中
- 第一批task执行完后,下一批task复用已有的shuffleFileGroup
磁盘文件数量 = core数量 * 下一个stage的task数量
SortShuffleManager运行原理
普通运行机制
- 数据先写入内存数据结构(reduceByKey—Map)
- 每写一条数据,判断是否达到某个临界值,达到则根据key对数据排序,再溢出到磁盘
- 写入磁盘通过java的BufferedOutputStream实现,先缓存到内存,满后一次写入磁盘文件
- 合并所有临时磁盘文件(merge),归并排序,依次写入同一个磁盘文件
- 单独写一份索引文件,标识下游各个task的数据在文件中的start and end
磁盘文件数量 = 上游stage的task数量
bypass运行机制
当reducer的task数量 < spark.sort.bypassMergeThreshold(200),shuffle write过程不排序,按未经优化的HashShuffleManager的方式write,最后将所有临时磁盘文件合并成一个,并创建索引文件