Spark shuffle调优

Posted 2021-01-27

Spark shuffle是什么
Shuffle在Spark中即是把父RDD中的KV对按照Key重新分区，从而得到一个新的RDD。也就是说原本同属于父RDD同一个分区的数据需要进入到子RDD的不同的分区。

现在的spark版本默认使用的是sortshuffle;
shuffle在哪里产生

shuffle在spark的算子中产生,也就是运行task的时候才会产生shuffle.

sortShuffleManagerspark

shuffle的默认计算引擎叫sortshuffleManager,它负责shuffle过程的执行、计算和组件的处理，sortshuffleManager会将task进行shuffle操作时产生的临时磁盘文件合并成一个磁盘文件，在下一个stage的shuffle read task拉取自己的数据时，只要根据索引读取每个磁盘文件中的部分数据即可。

sortshuffle的内部机制

1. 数据会根据不同的shuffle算子存储到map数据结构(如reduceByKey)或者array数据结构(join);不过Map是一边聚合,一边写入内存,array是直接写入内存. 当内存达到一个阈值,就会溢出写到磁盘,因此在溢出这个环节会在磁盘上产生多个临时文件,磁盘上的这些文件需要合并,于是spark就有了merge机制.

2. 在溢写到磁盘之前,在内存中会按照key来排序,排序过后会进入到一个buffer缓冲区,默认为32K,缓冲区的batch默认为1万条key,也就是缓冲区以每次一万条的量写入到磁盘文件中,该缓冲区减少IO,提高性能. 缓冲区和写入磁盘使用的技术是java中的BufferedOutputStream.

3. merge会将之前产生的所有的临时文件进行合并,包括缓冲区读写到磁盘上的文件,合并成一个大的文件到磁盘,默认为48M,与这个文件相对于的还有一个索引文件,索引文件里面记录的是这个文件的元信息,且这个磁盘文件也是下游stage的Task的输入信息! ?? 注: 一个下游的task对应一个磁盘文件和这个磁盘文件的元信息. 于是就有了血统,继承之类的!

shuffle当中可能会遇到的问题

1. 数据量非常大，从其他各台机器收集数据占用大量网络。
2. 数据如何分类，即如何Partition，Hash、Sort等；
3. 负载均衡（数据倾斜），因为采用不同的Shuffle方式对数据不同的分类，而分类之后又要跑到具体的节点上计算，如果不恰当的话，很容易产生数据倾斜；
4. 网络传输效率，需要在压缩和解压缩之间做出权衡，序列化和反序列也是要考虑的问题；
   说明：具体的Task进行计算的时候尽一切最大可能使得数据具备Process Locality的特性；退而求次是增加数据分片，减少每个Task处理的数据量。

shuffle调优
shuffle调优分为两种,一种是shuffle参数根据实际情况调优,一种是代码开发调优,代码开发调优我在spark性能调优里面去写!

1. spark.shuffle.file.buffer(默认值为32K,每次出货1万条)该参数是缓冲区的缓冲内存,如果可用的内存资源较为充足的话,可以将缓冲区的值设置大点,这样会较少磁盘IO次数.,如果合理调节该参数,性能会提升1%~5%...? 可以设置为64K.
2. spark.reducer.maxSizeInFlight(默认为48M)该参数是stage的每一个task就需要将上一个stage的计算结果中的所有相同key，从各个节点上通过网络都拉取到自己所在的节点上，然后进行key的聚合或连接等操作,如果合理调节该参数(增大),性能会提升1%~5%...
3. spark.shuffle.io.maxRetries(默认3次)该参数是stage的task向上一个stage的task计算结果拉取数据,也就是上面那个操作,有时候会因为网络异常原因,导致拉取失败,失败时候默认重新拉取三次,三次过还是失败的话作业就执行失败了,根据具体的业务可以考虑将默认值增大,这样可以避免由于JVM的一些原因或者网络不稳定等因素导致的数据拉取失败.也有助于提高spark作业的稳定性. 可以适当的提升重新拉取的次数,最大为60次.
4. spark.shuffle.io.retryWait(默认为5s)该参数和上面一样,是每次拉取数据的间隔时间...? 调优建议：建议加大间隔时长（比如20s），以增加shuffle操作的稳定性
5. spark.shuffle.memoryFraction(默认0.2,也就是20%)该参数是数据根据不同的shuffle算子将数据写入内存结构中,内存结构达到阈值会溢出临时文件,这个参数就是则是内存结构的阈值百分比的,不是内存结构的内存大小.? 如果内存充足，而且很少使用持久化操作，建议调高这个比例,可以减少频繁对磁盘进行IO操作,合理调节该参数可以将性能提升10%左右.
6. spark.shuffle.manager(默认sort)该参数是设置shuffle的类型,默认是sort,也就是sortshuffleManager, hash参数对应HashShuffleManager, tungsten-sort参数对应tungsten(这个很少用),HashShuffleManager是以前的版本,这个默认就行,
7. spark.shuffle.sort.bypassMergeThreshold(默认200个)该参数是如果shuffle read task的数量小于等于200个的时候,在sortshufflemanager模式下,就会启动ByPass sortshufflemanager...这个调优就这样把 ,默认200挺好的.
8. spark.shuffle.consolidateFiles(默认为false)该参数只对HashshuffleManager有效,而HashshuffleManager是spark1.2之前默认使用的版本...