Spark性能优化-------shuffle调优

影响一个Spark作业性能的因素，主要还是代码开发，资源参数以及数据倾斜，shuffle调优只能在整个Spark的性能调优中占到一小部分而已。

负责shuffle过程的执行，计算和处理的组件主要就是ShuffleManager。

SortShuffleManager运行原理
SortShuffleManager的运行机制主要分为两种，一种是普通运行机制，另一种是bypass运行机制，当shuffle
read task的数量小于等于spark.shuffle.sort.bypassMergeThreshold参数的值时（默认为200），就会启用bypass机制。
普通运行机制


X

1，根据不同的shuffle算子将数据写入内存数据结构中。每写一条数据进内存，就会判断是否达到临界阈值，如果达到，就spill到磁盘，然后清空内存数据结构。
2，spill到磁盘之前，会先对内存数据结构中已有的数据进行排序，分批写入磁盘文件。默认的batch数量是10000条。
3，一个task多次发生内存溢写磁盘的操作，产生多个临时文件，然后merge。一个task对应一个文件和一个索引文件。

Bypass运行机制

Bypass运行机制的触发条件：
1，shuffle map task数量小于spark.shuffle.sort.bypassMergeThreshold参数的值。
2，不是聚合类的shuffle算子（比如reduceByKey）。



简单来说，并行度低，产生文件数量少，没必要进行排序。
Task会为每个下游task都创建一个临时磁盘文件，将数据按key进行hash，将key 的hash值写入对应的磁盘文件之中，写入磁盘也是先写入内存缓冲，再spill到磁盘的方式。最后，同时会将所有临时磁盘文件都合并成一个磁盘文件，并创建一个单独的索引文件。
与普通SortShuffleManager运行机制不同在于，第一，磁盘写机制不同，第二，不会进行排序。Shuffle
write过程中，不需要进行数据的排序操作，节省掉这部分的性能开销。

Shuffle相关参数性能调优

Spark.shuffle.file.buffer
默认值：32k
参数说明：设置shuffle write task的BufferedOutputStream的buffer缓冲大小，将数据写到磁盘文件之前，就会写入buffer缓冲中，待缓冲写满之后，才会溢写到磁盘。
调优建议：内存足，调大（比如64k）。

spark.reducer.maxSizeInFlight
默认值：48m
参数说明：设置shuffle read task的buffer缓冲大小，这个buffer决定每次能够拉取多少数据。
调优建议：内存充足，调大（比如96m）。

spark.shuffle.io.maxRetries
默认值：3
参数说明：shuffle read task从shuffle write task所在节点拉取属于自己的数据时，如果因为网络异常导致拉取失败，就会自动进行重试，参数代表了重试的最大次数。
调优建议：对于超大数据量（数十亿~上百亿）的shuffle过程，调大该参数可大幅度提升稳定性。

spark.shuffle.io.retryWait
默认值：5s
参数说明：代表每次重试拉取数据的等待间隔，默认是5s。
调优建议：加大间隔时长（比如60s），以增加shuffle操作的稳定性。

spark.shuffle.memoryFracttion
默认值：0.2
参数说明：代表了Executor内存中，分配给shuffle read task进行聚合操作的内存比例，默认是20%。
调优建议：内粗足，且很少使用持久化操作，调高这个比例。避免由于内存不足导致聚合过程中频繁读写磁盘。
Spark.shuffle.managerr>
默认值：sort
参数说明：设置ShuffleManager的类型。有hash,sort,tungsten-sort可选。
调优建议：如果业务逻辑不需要对数据进行排序，可考虑其他机制来避免排序操作。

spark.shuffle.sort.bypassMergeThreshold
默认值：200
参数说明：当ShuffleManager为SortShuffleManager时，如果shuffle read task的数量小于这个阈值，则shuffle write过程中不会进行排序操作，按照未经优化的HashShufflemanager的方式去写数据。
调优建议：使用SortShuffleManager时，且不需要排序操作，将这个参数调大，大于shuffle read task的数量。

spark.shuffle.consilidateFiles
默认值：false
参数说明：如果使用HashShuffleManager，该参数有效。设置为true，就开启consolidate机制，会大幅度合并shuffle write的输出文件，对于shuffle
read task数量特别多的情况下，这种方法可极大地减少磁盘IO开销，提升性能。
调优建议：如果不需要SortShuffleManager的排序机制，除了使用bypass机制，还可以尝试将spark.shuffle.manager的参数手动指定为hash，使用HashShuffleManager,同时开启consolidate机制。