spark-shuffle

Shuffle就是对数据进行重组，由于分布式计算的特性和要求，在实现细节上更加繁琐和复杂
在MapReduce框架，Shuffle是连接Map和Reduce之间的桥梁，Map阶段通过shuffle读取数据并输出到对应的Reduce；而Reduce阶段负责从Map端拉取数据并进行计算。在整个shuffle过程中，往往伴随着大量的磁盘和网络I/O。所以shuffle性能的高低也直接决定了整个程序的性能高低。Spark也会有自己的shuffle实现过程。原文链接：https://blog.csdn.net/zhanglh046/article/details/78360762

总的来说，spark跟MR的shuffle并没有多大区别，都涉及到map(写数据的阶段)，跟reduce（读数据阶段）。

spark shuffle 执行流程

image.png

本文通过源码分析spark shuffle的执行过程，以及相关参数的调优。

通过分析spark 提交的源码，我们可以知道，最终调用的是org.apache.spark.scheduler.Task的runTask方法,而Task有2个子类，ShuffleMapTask（write（也可能存在先read后write,最后阶段是write）相当于MR中的Map阶段）跟ResultTask（开始阶段是read,相当于MR中的Reduce阶段）

大的流程shuffle-write需要经过write buffer、sort and spill、merge file三个阶段，细节上还是有差异，下面通过分析源码讲解shuffle-write的过程

image.png

shuffle write阶段

image.png

从ShuffleMapTask的runTask方法中查看SortShuffleManager的registerShuffle()

• 如果dep.mapSideCombine = false && numPartitions <= 200(spark.shuffle.sort.bypassMergeThreshold)，返回BypassMergeSortShuffleHandle

• 如果序列化器支持relocationserializer.supportsRelocationOfSerializedObjects && dep.mapSideCombine = false && numPartitions <= 1 << 24 // 16,777,216,（源码分析会解释为什么是<< 24位）返回SerializedShuffleHandle

• 其他情况返回BaseShuffleHandle

image.png

然后在getWriter方法中，根据registerShuffle返回的结果判断

• 如果是BypassMergeSortShuffleHandle返回BypassMergeSortShuffleWriter
• 如果是SerializedShuffleHandle返回UnsafeShuffleWriter
• 其他，即BaseShuffleHandle返回SortShuffleWriter

最后调用Writer的write 方法，所以，shuffle-write的具体流程分为BypassMergeSortShuffleWriter , UnsafeShuffleWriter , SortShuffleWriter三种write的流程

image.png

BypassMergeSortShuffleWriter 流程

image.png

BypassMergeSortShuffleWriter writer流程很简单

• 创建spark.shuffle.sort.bypassMergeThreshold(默认200)个DiskBlockObjectWriter对象，每个对象对应创建一个文件，文件名生成规则temp_shuffle_UUID。
• 根据数据key计算出往哪个DiskBlockObjectWriter,将数据序列化后往文件写数据。
• 最后会得到numPartitions个磁盘文件，再将这些文件合并成一个data文件（文件名"shuffle_" + shuffleId + "_" + mapId + "_0.data"）和一个index文件（文件名"shuffle_" + shuffleId + "_" + mapId + "_0.data"）
  
  image.png

这个Writer的优势是不需要排序，能够调整的参数

• spark.shuffle.sort.bypassMergeThreshold(默认200)
• 写文件的缓存大小spark.shuffle.file.buffer(默认32k)
• 序列化器spark.serializer(默认org.apache.spark.serializer.JavaSerializer)

可根据实际情况适当调整以上参数

UnsafeShuffleWriter 流程

image.png

image.png

image.png

image.png

• 数据通过序列化器写入到内存serBuffer中效果如下所示
  

  image.png

• 通过Unsafe将serBuffer数据copy到MemoryBlock page中，copy的时候会跳过前面16位，前16位不存储数据，具体还不清楚，只知道这是用Unsafe读取byte[]数据的操作，因为需要将数据copy到page中（堆外或者堆内内存），所以需要使用Unsafe操作

• 写入page的时候，前4或者8位存储数据的长度，构成以下数据结构

  
  
  image.png

• 在inmemorySort中记录数据的信息，用64位记录每条信息的数据，类似索引信息。其中，前24位记录partitionId，13位记录 page number的编号，最后27位记录数据在page中的偏移量。因为24位记录partitionId，所以前面使用该Writer的时候会判断numPartitions <= 1 << 24 // 16,777,216这个数就是因为用前24位来记录partitionId，所以不能超过这个数。后续通过page number信息跟offset信息可以找到数据

image.png

• spark用一张bit表描述page的使用情况，用PAGE_TABLE_SIZE = 1<<13 //8196标示使用的page情况，1代表使用，0代表未使用，为pages的使用情况作标记

• page通过判断是否使用堆外内存spark.memory.offHeap.enabled，开启堆外内存必须配置堆外内存大小spark.memory.offHeap.size，使用堆外内存的优势：https://cloud.tencent.com/developer/article/1513203

image.png

• 如果内存不足了，则会将数据排序，写磁盘，然后释放page，排序的时候，因为inmemorysort记录了每条数据的partitionId，所以只需要将”索引“排序就行，后续通过page number 和offset信息读取数据写入文件中

• 后续逻辑跟SortShuffleWriter差不多，将排好序的数据写入文件中，最后merge一个或者多个文和内存数据，形成一个大文件跟索引文件，具体流程将在下部分介绍SortShuffleWriter一起介绍

image.png

SortShuffleWriter 流程

前面介绍的2种Writer都有限制，可能会走上面2种Writer的算子有groupByKey、sortByKey、coalesce.最后的SortShuffleWriter像个万金油，使用无限制。

image.png

image.png

image.png

image.png

image.png

image.png

image.png

image.png

首先，shuffle write分为write buffer、spill disk、merge file 三个阶段，分别对应上图的1，2，3

• write buffer: 将数据写到缓存buffer中，如果容量大于0.7，则会申请扩容buffer，如果能够申请到内存，就扩容继续写buffer

• sort and spill: 内存不足的时候，就会把缓存的数据排序后spill 到磁盘中

• merge file: 经过1，2阶段后，会产生多个spill后的文件，以及buffer内存中的数据，将内存以及磁盘的数据，排序后合并成1个分区且排序的大文件（数据文件），跟一个索引文件（描述数据文件）,reducer会根据索引文件，拉取数据文件所需要的数据。
  

  image.png

writer buffer 阶段

需要map端聚合 PartitionedAppendOnlyMap

• spark没有使用jdk自带的map结构，而是使用一个数组实现map功能

• 计算key的hash值的位置pos，2pos，如果2pos位置没有数据，则在2pos的位置放入key，2pos +1 的位置放入value，如果2*pos上有值，计算key是否等于，如果等于，则用传入的函数更新，如wordcount中的reduceByKey(_ + _)， 计算出新的value后更新，如果不等于，则通过(pos + delta) & mask 的方法重新计算hash值得位置，delta 从1开始，遇到key存在每次递增1

• 每次插入后，会判断当前大约容量，通过估算得方式计算占用的内存，每32次估算一次，如果大于当前的内存，就会向taskMemoryManager.acquireExecutionMemory申请内存，如果申请成功，则继续写入，如果写入不成功，则spill磁盘，所以，第一个优化点，理论上executor内存越大，在内存可存储的数据越多，spill磁盘的次数越少，速度越快。 spill的过程，调用collection.destructiveSortedWritablePartitionedIterator(comparator)，首先会将数据往前移动，填满中间空缺的位置，然后将内存中的数据进行排序，用的排序算法是TimSort，最后按照分区且排序的形式写入文件中。
  

  image.png

不需要map端聚合 PartitionedPairBuffer

• spark没有使用jdk自带的list结构，而是使用一个数组实现list功能

• 功能比较简单，不需要mapCombine，只需要将数据按照kv追加到数组后面，spill溢写磁盘与PartitionedAppendOnlyMap一样，将数据排序后（分区且排序，不需要移动数据，填充空缺的位置，数据本身就是紧凑的）写入磁盘文件。

  
  
  image.png

shuffle read阶段

不管Writer经过了那些流程，最后都会产生一个分区排序的大文件，以及一个索引文件（描述大文件的分区信息）
源码查看ResultTask runTask，发现最后调用的是rdd.iterator(partition, context)，spark RDD执行通过递归迭代的方式(有点像new 对象的过程，都会先向上递归查看父类是否创建，spark中，子RDD都会查看父RDD是否计算或者缓存)。根据RDD链执行，通过getOrCompute，往前推，找到read的入口，ShuffleRdd的compute方法，找到ShuffleReader

image.png

image.png

查看read方法，所以重点就在读取数据（远程+本地）获取wrappedStreams这个阶段，即重点分析通过ShuffleBlockFetcherIterator获取数据的流程

image.png

ShuffleBlockFetcherIterator

本质上就是就是读取数据的阶段，你可以理解成，client向server发送请求，获取数据，每个reducer通过读取index文件，再去大文件（shuffle write阶段合并后且分区排序文件），读取所属自己分区的数据。主要是从client读取数据的过程，超时、并发度、异常重试等方面入手，server端则通过调整处理的并发数方面入手。

image.png

ShuffleBlockFetcherIterator.initialize

image.png

image.png

image.png

image.png

• 首先将val targetRequestSize = math.max(maxBytesInFlight / 5, 1L)，默认是48m/5=9.6m。判断拉取的数据是否大于9.6m或者一个address拉取的blocks数大于maxBlocksInFlightPerAddress（默认是Int.MaxValue）,所以只由9.6m控制，如果是，则封装成一个new FetchRequest(address, curBlocks)。如果每块数据特别小，那样一个请求需要跟很多台机器建立连接拉取数据，链接过多，read端压力大，可以通过调整spark.reducer.maxReqsInFlight的数量，限制每个read跟机器建立连接的数量，提高成功率

• 最后会封装成N个FetchRequest。然后开始遍历拉数据，判断是否isRemoteBlockFetchable，总之就是正在拉取的数据不能大于spark.reducer.maxSizeInFlight（默认48m）并且请求数不能超过maxReqsInFlight（Int.MaxValue）,不然就进入等待的deferredFetchRequests队列。所以，为了提高shuffle read的request并发读的数量，可以提高maxBytesInFlight（默认48m）的大小。并且单个address拉取的blocks数不能超过maxBlocksInFlightPerAddress（默认是Int.MaxValue），所以，降低maxBlocksInFlightPerAddress可以降低同时拉取的blocks数量，防止同时拉取多个blocks导致io过高，导致服务无响应、io超时等异常。

• 最终执行sendRequest请求拉取数据，调优参数spark.shuffle.io.maxRetries（默认是3）拉取失败重试的次数， spark.shuffle.io.retryWait（默认是5秒）失败后等待5秒后尝试重新拉取数据，可以根据实际情况，适当调高这2个参数，提高容错。

ExternalShuffleService

shuffle调优还有一点就是可以开启ExternalShuffleService 。
Spark 的 Executor 节点不仅负责数据的计算，还涉及到数据的管理。如果发生了 shuffle 操作，Executor 节点不仅需要生成 shuffle 数据，还需要负责处理读取请求。如果 一个 Executor 节点挂掉了，那么它也就无法处理 shuffle 的数据读取请求了，它之前生成的数据都没有意义了。

为了解耦数据计算和数据读取服务，Spark 支持单独的服务来处理读取请求。这个单独的服务叫做 ExternalShuffleService，运行在每台主机上，管理该主机的所有 Executor 节点生成的 shuffle 数据。有读者可能会想到性能问题，因为之前是由多个 Executor 负责处理读取请求，而现在一台主机只有一个 ExternalShuffleService 处理请求，其实性能问题不必担心，因为它主要消耗磁盘和网络，而且采用的是异步读取，所以并不会有性能影响。

解耦之后，如果 Executor 在数据计算时不小心挂掉，也不会影响 shuffle 数据的读取。而且Spark 还可以实现动态分配，动态分配是指空闲的 Executor 可以及时释放掉。

ExternalShuffleService本质是一个基于Netty写的Netty服务，所以相关调优就是对Netty参数的调优，主要有以下这些参数，具体调整，需要根据实际情况做出相应的调整。
spark.shuffle.io.serverThreads
spark.shuffle.io.receiveBuffer
spark.shuffle.io.backLog
spark.shuffle.io.sendBuffer

调优参数汇总

堆外内存相关

• spark.shuffle.io.preferDirectBufs:是否优先使用堆外内存
• spark.memory.offHeap.enabled: 是否启用堆外内存
• spark.memory.offHeap.size: 设置堆外内存大小

shuffle write阶段参数调优

• spark.executor.memory：通过分析write的过程中可以知道，单个task可用的内存越大，可申请的内存越大，spill disk的次数越少，速度越快，所以，可以适当提高该参数
• spark.sql.shuffle.partitions:提高并行度可以减少单个task处理的数据量，减少spill disk次数，降低oom风险，但是并不是越大越好，提高该参数，会增加task的数量，跟线程的数量一个道理，到达一定阈值，线程数的越多反而会增加系统上下文切换的压力，需要一点点测试，根据不同的任务，确定具体的数据
• spark.shuffle.file.buffer:默认32k， write spill磁盘的时候，缓冲区大小
• spark.shuffle.spill.batchSize: 默认10000，write spill磁盘的时候，默认一批写入的数量

shuffle read阶段参数调优

• spark.reducer.maxSizeInFlight:默认48m，一个请求拉取一个块的数据为48/5=9.6m,理想情况下会有5个请求同时拉数据，但是可能遇到一个大块，超过48m，就只有一个请求在拉数据，无法并行，所以可用适当提高该参数
• spark.reducer.maxReqsInFlight:shuffle read的时候最多有多少个请求同时拉取数据，默认是Integer.MAX_VALUE，一般不优化，不修改
• spark.reducer.maxBlocksInFlightPerAddress: 一个拉取的请求，包含多少个server，默认一个请求是9.6m，但是可能每个server拉取的文件非常小，只有几k，那样一个请求就需要请求上千个server拉取数据，容易导致超时等异常，所以，适当降低该参数
• spark.reducer.maxReqSizeShuffleToMem:read 过程中内存可以存放最大的数据量，超过将会把拉取的数据放到磁盘
• spark.shuffle.io.maxRetries：默认3次，一个请求拉取失败时重试次数，增大该参数，可能会延迟任务执行时间，但是可以提高任务成功率
• spark.shuffle.io.retryWait：默认5秒，一个请求拉取失败时的等待时间，增大该参数，可能会延迟任务执行时间，但是可以提高任务成功率
• spark.shuffle.io.clientThreads: 拉取数据client的线程个数, 可适当调高

调优效果

• 账单表28亿，优化前：4600秒，优化后：1200秒，耗时减少73.9%
• 订单表4.7亿，优化前：1600秒，优化后：720秒，耗时减少55%

image.png

T: 使用的内存 1T=1024G
P: 配置spark.sql.shuffle.partitions，1P=1000
C: cpu cores数量

参考链接
https://blog.csdn.net/zhanglh046/article/details/78360762
https://github.com/JerryLead/SparkInternals
https://www.cnblogs.com/itboys/p/9201750.html
https://www.dazhuanlan.com/2019/12/19/5dfb2a10d780d/
https://blog.csdn.net/pre_tender/article/details/101517789
https://www.bilibili.com/video/BV1sW41147vD?from=search&seid=12279554496967751348
https://www.jianshu.com/p/cda24891f9e4
https://cloud.tencent.com/developer/article/1513203