基于TungstenSort的Shuffle内核优化方案

基于Tungsten Sort的Shuffle实现机制的入口点仍然是SortShuffleManager类，与同样在SortShuffleManager类控制下的其他两种实现机制不同的是，基于Tungsten Sort的Shuffle实现机制使用的ShuffleHandle与ShuffleWriter分别为SerializedShuffleHandle与UnsafeShuffleWrit-er。因此对应的具体实现机制可以用图7-7来表示。

图7-7 基于TungstenSort的Shuffle实现机制的框架类图

在基于Sort的Shuffle内核一节中已经提到过SortShuffleManager根据内部采用的不同实现细节，分别给出两种排序模式，而基于Tungsten Sort的Shuffle实现机制对应的就是序列化排序模式。

从图7-7中可以看到，基于Sort的Shuffle实现机制，具体的写入器的选择与注册得到的ShuffleHandle类型有关，参考SortShuffleManager类的registerShuffle方法，相关代码如下。

在第15行代码处，会判断是否满足序列化模式的条件，如果满足，则使用基于Tung-stenSort的Shuffle实现机制，对应在代码中表现为使用类型为SerializedShuffleHandle的Shuf-fleHandle。上述代码进一步说明了在将“spark.shuffle.manager”设置为sort时，内部会自动选择具体的实现机制。对应代码的先后顺序就是选择的先后顺序。(www.xing528.com)

对应的序列化排序（Serialized sorting）模式需要满足的条件如下。

1）Shuffle依赖中不带聚合操作或没有对输出进行排序的要求。

2）Shuffle的序列化器支持序列化值的重定位（当前仅仅支持KryoSerializer及Spark SQL子框架自定义的序列化器）。

3）Shuffle过程中的输出分区个数少于16777216个。

实际上，使用过程中还有其他一些限制，比如由于使用Page形式的内存管理模型后，内部单条记录的长度不能超过128 MB（具体内存模型可以参考PackedRecordPointer类）。另外分区个数的限制也是该内存模型导致的（同样参考PackedRecordPointer类）。

所以，目前使用基于TungstenSort的Shuffle实现机制的条件还是比较苛刻的。