SortShuffleManager的bypass机制触发条件是什么？

SortShuffleManager的运行机制主要分成两种，一种是普通运行机制，另一种是bypass运行机制。当shuffle write。

task的数量小于等于spark.shuffle.sort.bypassMergeThreshold参数的值时(默认为200)，就会启用bypass机制。

该模式下，数据会先写入一个内存数据结构中(默认5M)，此时根据不同的shuffle算子，可能选用不同的数据结构。如果是reduceByKey这种聚合类的shuffle算子，那么会选用Map数据结构，一边通过Map进行聚合，一边写入内存;如果是join这种普通的shuffle算子，那么会选用Array数据结构，直接写入内存。

(2)接着，每写一条数据进入内存数据结构之后，就会判断一下，是否达到了某个临界阈值。如果达到临界阈值的话，那么就会尝试将内存数据结构中的数据溢写到磁盘，然后清空内存数据结构。

(3)排序

在溢写到磁盘文件之前，会先根据key对内存数据结构中已有的数据进行排序。

(4)溢写

排序过后，会分批将数据写入磁盘文件。默认的batch数量是10000条，也就是说，排序好的数据，会以每批1万条数据的形式分批写入磁盘文件。

(5)merge

一个task将所有数据写入内存数据结构的过程中，会发生多次磁盘溢写操作，也就会产生多个临时文件。最后会将之前所有的临时磁盘文件都进行合并成1个磁盘文件，这就是merge过程。由于一个task就只对应一个磁盘文件，也就意味着该task为Reduce端的stage的task准备的数据都在这一个文件中，因此还会单独写一份索引文件，其中标识了下游各个task的数据在文件中的start offset与end offset。

触发条件

bypass运行机制的触发条件如下： 1)shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold=200参数的值。 2)不是map combine聚合的shuffle算子(比如reduceByKey有map combie)。

bypass运行机制的触发条件如下：

1)shuffle map task数量小spark.shuffle.sort.bypassMergeThreshold=200参数的值。

2)不是map combine聚合的shuffle算子(比如reduceByKey有map combie)。

  • 此时task会为每个reduce端的task都创建一个临时磁盘文件，并将数据按key进行hash，然后根据key的hash值， 将key写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的 。最后，同样会将所有临时磁盘文件都合并成一个磁盘文件，并创建一个单独的索引文件。

  • 该过程的磁盘写机制其实跟未经优化的HashShuffleManager是一模一样的，因为都要创建数量惊人的磁盘文件， 只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件，也让该机制相对未经优化的 HashShuffleManager来说，shuffle read的性能会更好。

而该机制与普通SortShuffleManager运行机制的不同在于：

第一，磁盘写机制不同;

第二，不会进行排序。也就是说，启用该机制的最大好处在于，shuffle write过程中，不需要进行数据的排序操作， 也就节省掉了这部分的性能开销。

总结：

SortShuffle也分为普通机制和bypass机制

普通机制在内存数据结构(默认为5M)完成排序，会产生2M个磁盘小文件。

而当shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。或者算子不是聚合类的shuffle算子(比如reduceByKey)的时候会触发SortShuffle的bypass机制，SortShuffle的bypass机制不会进行排序，极大的提高了其性能。