SparkShuffle两种机制

1.HashShuffleManager(spark1.6之前)

普通的HashShuffleManager机制

普通HashShuffleManager机制

1)每个task有独立的buffer内存,根据reduce下一个stage 并行度的数量,产生3个(此处)buffer缓冲区;运算数据求hashcode对下一个stage并行度取余,进入对应的buffer缓冲区,buffer默认大小32k,满即溢写磁盘,一个buffer缓冲区对应一个blockfile磁盘文件;

2)所以map端的磁盘小文件的个数由 map端stage并行度和reduce端stage并行度决定;

缺点:

        小文件个数多,需要大量的IO时间;

        小文件过多,网络IO容易因为网络延迟出错;

         容易造成reduce端OOM;

优化的HashShuffleManager机制

优化HashShuffleManager机制

1) 优化的HashShuffleManager 串行的一组task共享一个buffer内存,对比原来,优化后,一个核只会产生对应下一个stage 并行度的buffer缓冲区,落地磁盘的小文件也少了许多;

2) 优化后,大大减少了map 端磁盘小文件的数量,减少了IO的压力

3) 但是当map端stage的核数和reduce stage的并行度都很高的时候,IO还是很费时间;

2.SortShuffle(spark1.6之后)

SortShuffle的运行机制主要分成两种： – 普通运行机制 – bypass运行机制

普通运行机制

SortShuffleManager  普通机制

1) :每一个task有一块大小为5M的内存缓冲区,缓冲区达到阈值时,落地磁盘过程中会排序聚合(即map端会有预聚合),最后落地到磁盘有两个磁盘文件,一个是数据文件,一个索引文件方便加快访问数据;

2) map 端磁盘小文件的数量为2倍的task数量;

3) reduce 端正常的shuffle read;

 bypass运行机制

SortShuffleManager  byPass机制

1) bypass 相对于普通的SortShuffleManager 机制,减少了map端的排序预聚合;

bypass机制开启条件:

        1.map端没有预聚合

        2.分区数要小于 spark.shuffle.sort.bypassMergeThreshold = 200