Spark从入门到精通10 -- Spark Shuffle_数据 混洗

Spark Shuffle

Shuffle简介

• Shuffle(数据混洗)是将一组无规则的数据转换为一组有规则的数据。Spark是一个分布式计算引擎，大多数的计算和数据转换过程是在多台计算机上执行的，当我们对RDD进行规约操作时,例如reduceByKey，或者当两个RDD之间是宽依赖的关系时，都会产生Shuffle。

Shuffle实现方案

• Shuffle过程中会导致RDD进行重分区，在数据量情况较大情况下也会将数据序列到磁盘上，所以整个Shuffle过程会有大量的网络通信和磁盘IO，Shuffle的效率高低成为了Spark计算效率高低的关键。

Hash Shuffle

• 在Spark1.6之前使用的Hash Shuffle

• Hash Shuffle原理图

• 过程解析

每一个MapTask会将数据按照key值进行hash然后按照ReduceTask需要的分区数进行分为指定个数的bucket，这里的bucket是内存缓存区，默认大小为100M，MapTask将数据平均的分布到每个bucket，当每个bucket快满了时(0.8)，将bucket内存中的数据写入到blockFile中，最终所有的数据都写入blockFile中，这里存在的性能问题是将产生大量的临时文件，临时文件个数为M*R，会产生大量的磁盘IO请求操作，效率低下。

• 过程解析

考虑到多个临时文件导致磁盘IO请求过多，产生效率问题。Hash Shuffle进行了优化。之前每个Executor中并行着多个Task任务，临时文件个数为M*R，现在考虑将每个Executor上执行的多个Task的结果写入到一个文件中，这样临时文件的个数为E*R，这个数量级比之前比小了并发Task数量倍数。但是如果Executor节点过多，效率还是不高。

Sort Based Shuffle

• Sort Based Shuffle原理图

• 过程解析

为了解决每个MapTask任务对应多个ReduceTask导致生成多个临时小文件的问题，Sort Based Shuffle引入了索引文件的概念。将每一个MapTask生成的对应每个分区的小bucket先合并成一个大的data文件，在data文件中采用segment(分段)的方式去记录中间数据，每个data文件对应一个index索引文件，用于标识每个分区起始位置和结束位置。磁盘io请求操作从变成了M数量级。

• 策略简介

1、SortShuffleWriter

SortShuffleWriter在将数据从临时bucket转化为大的data文件时会进行key值hash后排序(堆排)，并对相同的key的value值合并，这样生成的大data文件是一个有序的大文件。像reduceByKey这样的操作就使用这样的策略。

2、ByPassMergeSortShuffleWriter

ByPassMergeSortShuffleWriter在将数据从临时bucket文件转化为大的data文件时，不进行按照key值排序，也不会将相同key的value值进行合并。生成的data文件内部按照reduceId进行标识分段。因为这样写不在乎顺序所以都是并发写，但是当ReduceTask分区数过多时，会导致占用过多内存。这种混洗方式适用于ReduceTask数量不多(小于默认值200)并且不需要排序和map端聚合的任务效率更高。

3、UnasfeSortShuffleWriter

UnsafeShuffleWriter里面维护着一个 ShuffleExternalSorter， 用来做外部排序， 外部排序就是要先部分排序数据并把数据输出到磁盘，然后最后再进行merge全局排序。

和SortShuffleWriter的区别：

• 策略使用条件

1.在数据量小，Reduce分区数<=spark.shuffle.sort.by pass Merge Threshold( 默认200 ) 并且任务中没有Map端聚合操作时，采用类似Hase Based Shuffle的实现方式，不会对数据进行sort和merge，而是直接将每个Map Task的数据按Key值得到Reduce分区ID，然后每个分区生成一个DiskObjectWriter将该条数据Append到这个MapTask对应分区的临时文件里面，不同的是最后会对MapTask生成的临时文件进行合并，每个MapTask生成一个数据文件和一个索引文件。

2.当对象的序列化方式支持Recolation并且Reduce分区数小于2^24时，采用UnsafeSortshuffle，这种方式省去了将接收到的数据进行排序时要经历的反序列化然后再序列化的过程，直接将二进制数据写入MemoryBlock，同时将该数据的Reduce分区Id以及数据地址写入inMemorySorter的指针数组，排序时只对指针数组排序，spill时根据地址指针去获取数据。

3.最后一种SortShufleWriter，当Shuffle不满足上述两种条件时采用。内部采用ExternalSorter来溢写数据。其中，当没有map端聚合时，ExternalSorter内部采用PartitionedPairBuffer来接纳数据，这个时候即使job输出指定了排序逻辑，在shuffle write时并不会预排序，PartitionedPairBuffer的结构特点使得数据插入后是分区Id有序的，达到溢写磁盘条件时直接将Buffer内数据写出；当map端有聚合时，会采用PartitionedAppendOnlyMap来接纳数据，可以对key值进行跟新来完成聚合操作，并且这个时候job输出指定了排序逻辑的话，spill的时候数据会按照（Id,Key）的二元组进行排序。

逻辑摘抄自：
https://mp.weixin.qq.com/s?__biz=MzU3NTY3MTQzMg==&mid=100000768&idx=1&sn=0a5905b9f5567ea94f568120f1245c3c&chksm=7d1ed9e74a6950f1ad956d462a50c093ada8d588c5b499ca5a4501aacf3dcee1c88ae5d296a0&mpshare=1&scene=23&srcid=0214mymTPMCIplOiR6gfWO6U#rd
1
2
3
4
5
6
7
8

Shuffle中的数据倾斜

• 产生原因

在进行shuffle的时候，必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理，此时如果某个key对应的数据量特别大的话，就会发生数据倾斜。

• 解决方案

1、先局部再整体

先给每个key都打上一个随机数，比如10以内的随机数，此时原先一样的key就变成不一样的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就会变成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着对打上随机数后的数据，执行reduceByKey等聚合操作，进行局部聚合，那么局部聚合结果，就会变成了(1_hello, 2) (2_hello, 2)。然后将各个key的前缀给去掉，就会变成(hello,2)(hello,2)，再次进行全局聚合操作。将产生倾斜的数据分解，数据均匀分布到多个节点上，

2、替代join

将较小RDD中的数据直接通过collect拉取到Driver端的内存中来，然后对其创建一个Broadcast变量，接着对另外一个RDD执行map操作。