Spark Shuffle机制-源码实现

小咪咪 2022-10-12 12:16 116阅读 0赞

### . ###

*  一 .前言
 *  二 .Shuffle Write框架设计和实现
 *   *  2.1. BypassMergeSortShuffleWriter
     *  2.2. UnsafeShuffleWriter
     *  2.3. SortShuffleWriter
 *  三 . 输出文件验证
 *   *  3.1. 通用验证代码
     *   *  3.1.1 验证代码
         *  3.1.2 断点入口 : org.apache.spark.scheduler.ShuffleMapTask \# runTask
     *  3.2. SortShuffleWriter
     *   *  3.2.1 入口, 获取SortShuffleManager
         *  3.2.2 构建外部排序器: ExternalSorter
         *  3.2.3 插入数据
         *  3.2.4 判断溢写磁盘的类型
         *  3.1.5 验证溢写磁盘条件
         *  3.1.6 准备开始溢写磁盘
         *  3.1.7 溢写磁盘
         *  3.1.8 DiskBlockObjectWriter 写入器属性
         *  3.1.9 临时文件的生成规则
         *  3.1.10 准备合并 临时文件
         *  3.1.11 xxx
         *  3.1.12 xxx
         *  3.1.13 写入索引文件
         *  3.1.14 xxx
         *  3.1.15 xxx
     *  3.2. BypassMergeSortShuffleWriter
     *  3.3. UnsafeShuffleWriter
 *  四.Shuffle Read框架设计和实现

# 一 .前言 #

Shuffle机制分为Shuffle Write和Shuffle Read两个阶段，前者主要解决上游stage输出数据的分区问题，后者主要解决下游stage从上游stage获取数据、重新组织、并为后续操作提供数据的问题。

# 二 .Shuffle Write框架设计和实现 #

在Shuffle Write阶段，数据操作需要分区、聚合和排序3个功能但每个数据操作只需要其中的⼀个或两个功能。

Spark为了⽀持所有的情况，设计了⼀个通⽤的Shuffle Write框架，框架的计算顺序为“map()输出→数据聚合→排序→分区”输出

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

如上图所示, map task每计算出⼀个record及其partitionId，就将record放⼊类似HashMap的数据结构中进⾏聚合；聚合完成后，再将HashMap中的数据放⼊类似Array的数据结构中进⾏排序，既可按照partitionId，也可以按照partitionId+Key进⾏排序；最后根据partitionId将数据写⼊不同的数据分区中，存放到本地磁盘上。其中，聚合（aggregate，即combine）和排序（sort）过程是可选的，如果数据操作不需要聚合或者排序，那么可以去掉相应的聚合或排序过程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

*  Shuffle Write 一共有三种实现方式 :

<table> 
 <thead> 
  <tr> 
   <th align="left">名称</th> 
   <th align="left">是否支持map端预聚合</th> 
   <th align="left">是否支持排序</th> 
   <th align="left">限制条件</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">BypassMergeSortShuffleWriter</td> 
   <td align="left">否</td> 
   <td align="left">否</td> 
   <td align="left">1. shuffle reduce task 数量小于等于 spark.shuffle.sort.bypassMergeThreshold 参数的值，默认为 200。<br> 2.不是聚合类的 shuffle 算子（比如 reduceByKey）。</td> 
  </tr> 
  <tr> 
   <td align="left">UnsafeShuffleWriter</td> 
   <td align="left">否</td> 
   <td align="left">否</td> 
   <td align="left">1.是否支持序列化重定位 , 默认java是不支持, kyro支持. <br> 2. map端不支持 预聚合操作. <br> 3. 下游分区的数量不能大于 1677 7215 + 1 (一千六百多万分区)</td> 
  </tr> 
  <tr> 
   <td align="left">SortShuffleWriter [默认]</td> 
   <td align="left">是 ( 看使用的算子是否启用聚合)</td> 
   <td align="left">是 <br> (如果支持map端预聚合功能 (排序) 否则不排序 )</td> 
   <td align="left">默认实现, 支持map端聚合,支持排序</td> 
  </tr> 
 </tbody> 
</table>

## 2.1. BypassMergeSortShuffleWriter ##

> 触发机制:
> 
> 1.  shuffle reduce task 数量小于等于 spark.shuffle.sort.bypassMergeThreshold 参数的值，默认为 200。
> 2.  不是聚合类的 shuffle 算子（比如 reduceByKey

task 会为每个 reduce 端的 task 都创建一个临时磁盘文件，并将数据按 key 进行hash 然后根据 key 的 hash 值，将 key 写入对应的磁盘文件之中。当然，写入磁盘文件时也是先写入内存缓冲，缓冲写满之后再溢写到磁盘文件的。最后，同样会将所有临时磁盘文件  
都合并成一个磁盘文件，并创建一个单独的索引文件

该过程的磁盘写机制其实跟未经优化的 HashShuffleManager 是一模一样的，因为都要创建数量惊人的磁盘文件，只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件，也让该机制相对未经优化的 HashShuffleManager 来说， shuffle read 的性能会更好。  
而该机制与普通 SortShuffleManager 运行机制的不同在于：不会进行排序。也就是说，启用该机制的最大好处在于， **shuffle write 过程中，不需要进行数据的排序操作，也就节省掉了这部分的性能开销**

*  触发条件代码片段: **SortShuffleWriter.shouldBypassMergeSort**

def shouldBypassMergeSort(conf: SparkConf, dep: ShuffleDependency[_, _, _]): Boolean = { 
    
    
        // 是否支持map端预聚合
        // We cannot bypass sorting if we need to do map-side aggregation.
        if (dep.mapSideCombine) { 
          require(dep.aggregator.isDefined, "Map-side combine without Aggregator specified!")
          false
        } else { 
          // 下游分区数小于200
          val bypassMergeThreshold: Int = conf.getInt("spark.shuffle.sort.bypassMergeThreshold", 200)
          dep.partitioner.numPartitions <= bypassMergeThreshold
        }
      }

## 2.2. UnsafeShuffleWriter ##

> 触发机制:
> 
> 1.  序列化规则支持重定位操作(java序列化不支持, kryo支持.)。
> 2.  不是聚合类的 shuffle 算子（比如 reduceByKey)
> 3.  下游分区的数量不能大于 1677 7215 + 1 (一千六百多万分区)

*  触发条件代码片段: **SortShuffleManager.canUseSerializedShuffle**

/** * Helper method for determining whether a shuffle should use an optimized serialized shuffle * path or whether it should fall back to the original path that operates on deserialized objects. */
      def canUseSerializedShuffle(dependency: ShuffleDependency[_, _, _]): Boolean = { 
        val shufId = dependency.shuffleId
        val numPartitions = dependency.partitioner.numPartitions
    
        // 是否支持序列化重定位 , 默认java是不支持, kyro支持.
        if (!dependency.serializer.supportsRelocationOfSerializedObjects) { 
          log.debug(s"Can't use serialized shuffle for shuffle $shufId because the serializer, " +
            s"${dependency.serializer.getClass.getName}, does not support object relocation")
          false
        // map端不支持 预聚合操作.
        } else if (dependency.aggregator.isDefined) { 
          log.debug(
            s"Can't use serialized shuffle for shuffle $shufId because an aggregator is defined")
          false
          // 下游分区的数量不能大于 1677 7215 + 1 (一千六百多万分区)
        } else if (numPartitions > MAX_SHUFFLE_OUTPUT_PARTITIONS_FOR_SERIALIZED_MODE) { 
          log.debug(s"Can't use serialized shuffle for shuffle $shufId because it has more than " +
            s"$MAX_SHUFFLE_OUTPUT_PARTITIONS_FOR_SERIALIZED_MODE partitions")
          false
        } else { 
          log.debug(s"Can use serialized shuffle for shuffle $shufId")
          true
        }
      }

## 2.3. SortShuffleWriter ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

在该模式下，数据会先写入一个数据结构， reduceByKey 写入 Map，一边通过 Map 局部聚合，一遍写入内存。 Join 算子写入 ArrayList 直接写入内存中。然后需要判断是否达到阈值，如果达到就会将内存数据结构的数据写入到磁盘，清空内存数据结构。

在溢写磁盘前，先根据 key 进行排序，排序过后的数据，会分批写入到磁盘文件中。

写入磁盘文件通过缓冲区溢写的方式，每次溢写都会产生一个磁盘文件， 也就是说一个 Task 过程会产生多个临时文件。  
最后在每个 Task 中，将所有的临时文件合并，这就是 merge 过程，此过程将所有临时文件读取出来，一次写入到最终文件。 意味着一个 Task 的所有数据都在这一个文件中。同时单独写一份索引文件，标识下游各个Task的数据在文件中的索引， start offset和end offset。

*  写入源码 SortShuffleWriter\#write

/** Write a bunch of records to this task's output */
      override def write(records: Iterator[Product2[K, V]]): Unit = {
    
    
        // 1. 获取排序器
        sorter = if (dep.mapSideCombine) {
          require(dep.aggregator.isDefined, "Map-side combine without Aggregator specified!")
          new ExternalSorter[K, V, C](
            context, dep.aggregator, Some(dep.partitioner), dep.keyOrdering, dep.serializer)
        } else {
          // In this case we pass neither an aggregator nor an ordering to the sorter, because we don't
          // care whether the keys get sorted in each partition; that will be done on the reduce side
          // if the operation being run is sortByKey.
          new ExternalSorter[K, V, V](
            context, aggregator = None, Some(dep.partitioner), ordering = None, dep.serializer)
        }
    
        // 2. 所有的数据放入排序器. 这个过程中 根据阈值溢写到磁盘.
        sorter.insertAll(records)
    
        // Don't bother including the time to open the merged output file in the shuffle write time,
        // because it just opens a single file, so is typically too fast to measure accurately
        // (see SPARK-3570).
    
    
        // 打开一个文件, 写入数据.
        val output = shuffleBlockResolver.getDataFile(dep.shuffleId, mapId)
    
        // 生成文件名称
        val tmp = Utils.tempFileWith(output)
        try {
    
          logInfo("溢写数据文件.....")
          // 这一步将溢写到的磁盘的文件和内存中的数据进行归排序
          // 并溢写到一个文件中,这一步写的文件是临时的文件名
          val blockId = ShuffleBlockId(dep.shuffleId, mapId, IndexShuffleBlockResolver.NOOP_REDUCE_ID)
          val partitionLengths = sorter.writePartitionedFile(blockId, tmp)
    
    
          // 溢写索引文件
          logInfo("溢写索引文件.....")
          shuffleBlockResolver.writeIndexFileAndCommit(dep.shuffleId, mapId, partitionLengths, tmp)
    
          mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths)
        } finally {
          if (tmp.exists() && !tmp.delete()) {
            logError(s"Error while deleting temp file ${tmp.getAbsolutePath}")
          }
        }
      }

*  溢写条件 Spillable\#maybeSpill

> 1.  申请不到足够的内存( 默认初始内存是5M)
> 2.  数据条数超过限制 `spark.shuffle.spill.numElementsForceSpillThreshold` 默认Long.MaxValue

/** * Spills the current in-memory collection to disk if needed. Attempts to acquire more * memory before spilling. * * @param collection collection to spill to disk * @param currentMemory estimated size of the collection in bytes * @return true if `collection` was spilled to disk; false otherwise */
      protected def maybeSpill(collection: C, currentMemory: Long): Boolean = {
        var shouldSpill = false
    
        // 元素是32的整数倍 并且 当前使用的内存 大于 初始内存的阈值(5M)
        if (elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold) {
    
          // 从shuffle内存池中获取最多两倍的当前内存
          // Claim up to double our current memory from the shuffle memory pool
          val amountToRequest = 2 * currentMemory - myMemoryThreshold
          // 申请内存
          val granted = acquireMemory(amountToRequest)
    
          // 更新当前使用的内存
          myMemoryThreshold += granted
    
          // 如果新申请的内存还放不下数据的话, 则直接溢写
          // If we were granted too little memory to grow further (either tryToAcquire returned 0,
          // or we already had more memory than myMemoryThreshold), spill the current collection
          shouldSpill = currentMemory >= myMemoryThreshold
        }
        // 是否溢写 : 是否超过内存限制 或者 数据条数大于 限制 Long.MaxValue .
        shouldSpill = shouldSpill || _elementsRead > numElementsForceSpillThreshold
    
        // 开始执行溢写操作
        // Actually spill
        if (shouldSpill) {
          _spillCount += 1
          // 记录日志
          logSpillage(currentMemory)
    
          // 溢写数据
          spill(collection)
    
          // 初始化
          _elementsRead = 0
          _memoryBytesSpilled += currentMemory
    
          // 释放内存
          releaseMemory()
        }
        shouldSpill
      }

# 三 . 输出文件验证 #

主要为了验证 writer写出文件的过程

## 3.1. 通用验证代码 ##

### 3.1.1 验证代码 ###

import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import scala.Tuple2;
    
    import java.io.File;
    
    
    public class SparkShuffleFileTest { 
      public static void main(String[] args) { 
    
        String path = "/opt/a/spark_test/shuffle.txt" ;
    
        String resPath = "file:///opt/a/spark_test/res";
        File res = new File(resPath);
        res.deleteOnExit();
    
        SparkConf conf = new SparkConf().setAppName("app").setMaster("local[4]");
    	// 每1000条数据 溢写一次磁盘...
        conf.set("spark.shuffle.spill.numElementsForceSpillThreshold","1000");
    
        JavaSparkContext sc  = new JavaSparkContext(conf);
    
        JavaRDD<String> textFile = sc.textFile(path,4);
        JavaPairRDD<String,Integer> data = textFile.mapToPair(s -> new Tuple2<String,Integer>(s, 1)).reduceByKey((a,b) -> a + b);
    
        data.groupByKey().saveAsTextFile(resPath);
    
    
      }
    
    }

### 3.1.2 断点入口 : org.apache.spark.scheduler.ShuffleMapTask \# runTask ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

## 3.2. SortShuffleWriter ##

### 3.2.1 入口, 获取SortShuffleManager ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]

### 3.2.2 构建外部排序器: ExternalSorter ###

org.apache.spark.shuffle.sort.SortShuffleWriter\#write

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 5]

### 3.2.3 插入数据 ###

def insertAll(records: Iterator[Product2[K, V]]): Unit = { 
        // TODO: stop combining if we find that the reduction factor isn't high
    
        // 是否开启预聚合
        val shouldCombine = aggregator.isDefined
    
        // 是否开启预聚合
        if (shouldCombine) { 
          // 开启
          // Combine values in-memory first using our AppendOnlyMap
          val mergeValue = aggregator.get.mergeValue
          val createCombiner = aggregator.get.createCombiner
          var kv: Product2[K, V] = null
          val update = (hadValue: Boolean, oldValue: C) => { 
            if (hadValue) mergeValue(oldValue, kv._2) else createCombiner(kv._2)
          }
          while (records.hasNext) { 
            addElementsRead()
            kv = records.next()
    
            // 向内存缓冲 中 插入一条数据, map的键值为 ( partitionId , key )
            map.changeValue((getPartition(kv._1), kv._1), update)
    
            // 是否触发阈值, 溢写数据到磁盘
            maybeSpillCollection(usingMap = true)
          }
        } else { 
          // Stick values into our buffer
          while (records.hasNext) { 
            addElementsRead()
            val kv = records.next()
            buffer.insert(getPartition(kv._1), kv._1, kv._2.asInstanceOf[C])
            maybeSpillCollection(usingMap = false)
          }
        }
      }

### 3.2.4 判断溢写磁盘的类型 ###

通过`org.apache.spark.util.collection.ExternalSorter#maybeSpillCollection` 进行判断是否要溢写磁盘  
map和buffer类型的处理逻辑不一样, 正常开启map端聚合都是使用 `usingMap= true`

/** * Spill the current in-memory collection to disk if needed. * * @param usingMap whether we're using a map or buffer as our current in-memory collection */
      private def maybeSpillCollection(usingMap: Boolean): Unit = { 
        var estimatedSize = 0L
        if (usingMap) { 
          estimatedSize = map.estimateSize()
          // 是否需要溢写
          // 1. 申请不到足够的内存
          // 2. 数据条数超过限制,默认Long.MaxValue 不起作用
          if (maybeSpill(map, estimatedSize)) { 
            map = new PartitionedAppendOnlyMap[K, C]
          }
        } else { 
          // 非map类型的数据
          estimatedSize = buffer.estimateSize()
          if (maybeSpill(buffer, estimatedSize)) { 
            buffer = new PartitionedPairBuffer[K, C]
          }
        }
    
        if (estimatedSize > _peakMemoryUsedBytes) { 
          _peakMemoryUsedBytes = estimatedSize
        }
      }

### 3.1.5 验证溢写磁盘条件 ###

通过`org.apache.spark.util.collection.ExternalSorter#maybeSpill` 进行判断是否要溢写磁盘

/** * Spills the current in-memory collection to disk if needed. Attempts to acquire more * memory before spilling. * * @param collection collection to spill to disk * @param currentMemory estimated size of the collection in bytes * @return true if `collection` was spilled to disk; false otherwise */
      protected def maybeSpill(collection: C, currentMemory: Long): Boolean = { 
        var shouldSpill = false
    
        // 元素是32的整数倍 并且 当前使用的内存 大于 初始内存的阈值(5M)
        if (elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold) { 
    
          // 从shuffle内存池中获取最多两倍的当前内存
          // Claim up to double our current memory from the shuffle memory pool
          val amountToRequest = 2 * currentMemory - myMemoryThreshold
          // 申请内存
          val granted = acquireMemory(amountToRequest)
    
          // 更新当前使用的内存
          myMemoryThreshold += granted
    
          // 如果新申请的内存还放不下数据的话, 则直接溢写
          // If we were granted too little memory to grow further (either tryToAcquire returned 0,
          // or we already had more memory than myMemoryThreshold), spill the current collection
          shouldSpill = currentMemory >= myMemoryThreshold
        }
        // 是否溢写 : 是否超过内存限制 或者 数据条数大于`spark.shuffle.spill.numElementsForceSpillThreshold` 限制 Long.MaxValue .
        shouldSpill = shouldSpill || _elementsRead > numElementsForceSpillThreshold
    
        // 开始执行溢写操作
        // Actually spill
        if (shouldSpill) { 
          _spillCount += 1
          // 记录日志
          logSpillage(currentMemory)
    
          // 溢写数据
          spill(collection)
    
          // 初始化
          _elementsRead = 0
          _memoryBytesSpilled += currentMemory
    
          // 释放内存
          releaseMemory()
        }
        shouldSpill
      }

### 3.1.6 准备开始溢写磁盘 ###

通过`org.apache.spark.util.collection.ExternalSorter#spill` 进行判断是否要溢写磁盘

溢写磁盘的时候首先要排序,然后再写入文件.  
溢写后的文件存放在集合 private val spills = new ArrayBuffer\[SpilledFile\] 里面…

/** * Spill our in-memory collection to a sorted file that we can merge later. * We add this file into `spilledFiles` to find it later. * * @param collection whichever collection we're using (map or buffer) */
      override protected[this] def spill(collection: WritablePartitionedPairCollection[K, C]): Unit = { 
    
        // 排序
        val inMemoryIterator = collection.destructiveSortedWritablePartitionedIterator(comparator)
    
        // 溢写文件
        val spillFile = spillMemoryIteratorToDisk(inMemoryIterator)
        spills += spillFile
      }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 6]

### 3.1.7 溢写磁盘 ###

将数据溢写到磁盘 .

/** * Spill contents of in-memory iterator to a temporary file on disk. */
      private[this] def spillMemoryIteratorToDisk(inMemoryIterator: WritablePartitionedIterator)
          : SpilledFile = { 
    
    
        // 创建临时文件
        // Because these files may be read during shuffle, their compression must be controlled by
        // spark.shuffle.compress instead of spark.shuffle.spill.compress, so we need to use
        // createTempShuffleBlock here; see SPARK-3426 for more context.
        val (blockId, file) = diskBlockManager.createTempShuffleBlock()
    
        // 这些变量在每次刷新后重置
        // These variables are reset after each flush
        var objectsWritten: Long = 0
        val spillMetrics: ShuffleWriteMetrics = new ShuffleWriteMetrics
    
        // 获取磁盘的溢写对象
        // fileBufferSize : 32k
    
        // writer = {DiskBlockObjectWriter@7500}
        // file = {File@7473} "/private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-021881f2-19bd-4981-8c78-c4fc8136f8ae/06/temp_shuffle_bae24a5b-22a4-4034-89fd-d726a283ee01"
        // serializerManager = {SerializerManager@7073}
        // serializerInstance = {JavaSerializerInstance@7375}
        // bufferSize = 32768
        // syncWrites = false
        // writeMetrics = {ShuffleWriteMetrics@7487}
        // blockId = {TempShuffleBlockId@7472} "temp_shuffle_bae24a5b-22a4-4034-89fd-d726a283ee01"
        // channel = null
        // mcs = null
        // bs = null
        // fos = null
        // ts = null
        // objOut = null
        // initialized = false
        // streamOpen = false
        // hasBeenClosed = false
        // committedPosition = 0
        // reportedPosition = 0
        // numRecordsWritten = 0
        // log_ = null
        val writer: DiskBlockObjectWriter =
          blockManager.getDiskWriter(blockId, file, serInstance, fileBufferSize, spillMetrics)
    
        // List of batch sizes (bytes) in the order they are written to disk
        val batchSizes = new ArrayBuffer[Long]
    
        // 按写入磁盘顺序排列的批大小（字节）列表
        // How many elements we have in each partition
        val elementsPerPartition = new Array[Long](numPartitions)
    
        // Flush the disk writer's contents to disk, and update relevant variables.
        // The writer is committed at the end of this process.
        def flush(): Unit = { 
          val segment = writer.commitAndGet()
          batchSizes += segment.length
          _diskBytesSpilled += segment.length
          objectsWritten = 0
        }
    
        var success = false
        try { 
          while (inMemoryIterator.hasNext) { 
            // 获取分区id
            val partitionId = inMemoryIterator.nextPartition()
            require(partitionId >= 0 && partitionId < numPartitions,
              s"partition Id: ${partitionId} should be in the range [0, ${numPartitions})")
            // 写入数据
            inMemoryIterator.writeNext(writer)
            elementsPerPartition(partitionId) += 1
            objectsWritten += 1
    
            if (objectsWritten == serializerBatchSize) { 
              flush()
            }
          }
          if (objectsWritten > 0) { 
            flush()
          } else { 
            writer.revertPartialWritesAndClose()
          }
          success = true
        } finally { 
          if (success) { 
            writer.close()
          } else { 
            // This code path only happens if an exception was thrown above before we set success;
            // close our stuff and let the exception be thrown further
            writer.revertPartialWritesAndClose()
            if (file.exists()) { 
              if (!file.delete()) { 
                logWarning(s"Error deleting ${file}")
              }
            }
          }
        }
        // 封装返回的SpilledFile 对象,记录数据
        // 15 = {ExternalSorter$SpilledFile@7820} "SpilledFile(/private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-021881f2-19bd-4981-8c78-c4fc8136f8ae/05/temp_shuffle_224b1b0f-ef98-46f2-92a2-0c5175a95b43,temp_shuffle_224b1b0f-ef98-46f2-92a2-0c5175a95b43,[J@41af1a93,[J@65ecae5e)"
        // file = {File@7914} "/private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-021881f2-19bd-4981-8c78-c4fc8136f8ae/05/temp_shuffle_224b1b0f-ef98-46f2-92a2-0c5175a95b43"
        // blockId = {TempShuffleBlockId@7915} "temp_shuffle_224b1b0f-ef98-46f2-92a2-0c5175a95b43"
        // serializerBatchSizes = {long[1]@7916} [4383]
        // elementsPerPartition = {long[4]@7917} [251, 251, 249, 249]
        // $outer = {ExternalSorter@7399}
        SpilledFile(file, blockId, batchSizes.toArray, elementsPerPartition)
      }

我们看到溢写到磁盘的时候是有序列化&压缩操作. 所以写入的数据文件都是二进制的文件. 这个没办法查看.

数据写入的格式 为 K-V

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 7]

### 3.1.8 DiskBlockObjectWriter 写入器属性 ###

总结一下 :

1.  采用的序列化工具为默认的 JavaSerializer
2.  压缩方式默认采用LZ4

this = { DiskBlockObjectWriter@7981} 
    	 file = { File@7978} "/private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-021881f2-19bd-4981-8c78-c4fc8136f8ae/1b/temp_shuffle_f1835a20-8b4b-4c00-b6e3-1afb46b7d30c"
    	 serializerManager = { SerializerManager@7073} 
    	 serializerInstance = { JavaSerializerInstance@7074} 
    	 bufferSize = 32768
    	 syncWrites = false
    	 writeMetrics = { ShuffleWriteMetrics@7980} 
    	 blockId = { TempShuffleBlockId@7977} "temp_shuffle_f1835a20-8b4b-4c00-b6e3-1afb46b7d30c"
    	 channel = { FileChannelImpl@8101} 
    	 mcs = { DiskBlockObjectWriter$ManualCloseBufferedOutputStream$1@8102} 
    	 bs = { LZ4BlockOutputStream@8103} "LZ4BlockOutputStream(out=org.apache.spark.storage.DiskBlockObjectWriter$ManualCloseBufferedOutputStream$1@4047a5ad, blockSize=32768, compressor=LZ4JNICompressor, checksum=StreamingXXHash32JNI(seed=-1756908916))"
    	 fos = { FileOutputStream@8104} 
    	 ts = { TimeTrackingOutputStream@8105} 
    	 objOut = { JavaSerializationStream@8106} 
    	 initialized = true
    	 streamOpen = true
    	 hasBeenClosed = false
    	 committedPosition = 0
    	 reportedPosition = 0
    	 numRecordsWritten = 2
    	 log_ = null

### 3.1.9 临时文件的生成规则 ###

在 temp\_shuffle的父目录,默认是 64个. 由参数 `spark.diskStore.subDirectories` 控制.  
目录的名称为16进制的数字, 名称长度固定为2位.

tmp\_shuffle的文件名称为 tmp\_shuffle\_uuid , 存放规则为 `hash % spark.diskStore.subDirectories` .

/private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-a615766f-d08b-4868-9ab3-b4718f1afaf7
    blockmgr
    
    
    - 工作目录 : 
    	|
    	|- blockmgr-a615766f-d08b-4868-9ab3-b4718f1afaf7   
    		|
    		|- 01  子目录 (一共64个, 由参数 `spark.diskStore.subDirectories`  控制.  目录的名称为16进制的数字,  名称长度固定为2位.  )
    				|
    				|- temp_shuffle_xxxxxxxxx
    				|- temp_shuffle_xxxxxxxxx
    		|- 0f 
    				|- temp_shuffle_xxxxxxxxx
    		|- xxx

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 8]

*  生成 临时的 shuffle文件

/** Produces a unique block id and File suitable for storing shuffled intermediate results. */
      def createTempShuffleBlock(): (TempShuffleBlockId, File) = { 
    
        // blockId实例 : temp_shuffle_f333d690-deaf-4b17-8b30-c655f9616937
        var blockId = new TempShuffleBlockId(UUID.randomUUID())
        while (getFile(blockId).exists()) { 
          blockId = new TempShuffleBlockId(UUID.randomUUID())
        }
        // 返回: /private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-021881f2-19bd-4981-8c78-c4fc8136f8ae/0a/temp_shuffle_f333d690-deaf-4b17-8b30-c655f9616937
        (blockId, getFile(blockId))
      }

生成文件名称

/** Looks up a file by hashing it into one of our local subdirectories. */
      // This method should be kept in sync with
      // org.apache.spark.network.shuffle.ExternalShuffleBlockResolver#getFile().
      def getFile(filename: String): File = { 
        // filename : temp_shuffle_402bcc43-e1d1-44c2-9e5b-53d7f6f62ceb
    
    
        // Figure out which local directory it hashes to, and which subdirectory in that
    
        // 1340714088
        val hash = Utils.nonNegativeHash(filename)
    
        // 0
        val dirId = hash % localDirs.length
    
        // 40
        val subDirId = (hash / localDirs.length) % subDirsPerLocalDir
    
        // Create the subdirectory if it doesn't already exist
        val subDir = subDirs(dirId).synchronized { 
          val old = subDirs(dirId)(subDirId)
          if (old != null) { 
            old
          } else { 
            // old = 0 ;
    
            // tmp_shuffle的上一级目录不存在 , 需要根据subDirId格式为16进制的两位数字, 创建.
            // 目录的名称由参数 `spark.diskStore.subDirectories` 控制
            val newDir = new File(localDirs(dirId), "%02x".format(subDirId))
            if (!newDir.exists() && !newDir.mkdir()) { 
              throw new IOException(s"Failed to create local dir in $newDir.")
            }
            subDirs(dirId)(subDirId) = newDir
    
            // /private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-021881f2-19bd-4981-8c78-c4fc8136f8ae/28
            newDir
          }
        }
    
        new File(subDir, filename)
      }

### 3.1.10 准备合并 临时文件 ###

org.apache.spark.shuffle.sort.SortShuffleWriter\#write

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 9]

/** * Write all the data added into this ExternalSorter into a file in the disk store. This is * called by the SortShuffleWriter. * * @param blockId block ID to write to. The index file will be blockId.name + ".index". * @return array of lengths, in bytes, of each partition of the file (used by map output tracker) */
      def writePartitionedFile(
          blockId: BlockId,
          outputFile: File): Array[Long] = { 
    
    
          // xxxx/blockmgr-9c480511-104d-48ac-b2cd-ac8dacbadfca/0c/shuffle_0_0_0.data.a9312076-d7ca-456c-9e1c-62b18ae348b3
        logInfo(" data 文件路径: ExternalSorter # writePartitionedFile "+ outputFile.getAbsolutePath)
    
        // Track location of each range in the output file
        // 获取分区的数量 4 , 用于记录偏移量 , 下标即为stage的 偏移量
        val lengths = new Array[Long](numPartitions)
    
        // 构建 Writer
        // writer = {DiskBlockObjectWriter@7180}
        // file = {File@7140} "/private/var/folders/37/9746t_yx10v2g49vkwtzjw_80000gn/T/blockmgr-9c480511-104d-48ac-b2cd-ac8dacbadfca/0c/shuffle_0_0_0.data.a9312076-d7ca-456c-9e1c-62b18ae348b3"
        // serializerManager = {SerializerManager@2827}
        // serializerInstance = {JavaSerializerInstance@7123}
        // bufferSize = 32768
        // syncWrites = false
        // writeMetrics = {ShuffleWriteMetrics@7115}
        // blockId = {ShuffleBlockId@7148} "shuffle_0_0_0"
        // channel = null
        // mcs = null
        // bs = null
        // fos = null
        // ts = null
        // objOut = null
        // initialized = false
        // streamOpen = false
        // hasBeenClosed = false
        // committedPosition = 0
        // reportedPosition = 0
        // numRecordsWritten = 0
        // log_ = null
        val writer = blockManager.getDiskWriter(blockId, outputFile, serInstance, fileBufferSize,
          context.taskMetrics().shuffleWriteMetrics)
    
        if (spills.isEmpty) { 
          // Case where we only have in-memory data
          val collection = if (aggregator.isDefined) map else buffer
          val it = collection.destructiveSortedWritablePartitionedIterator(comparator)
    
          // 处理额内存中的数据
          while (it.hasNext) { 
            val partitionId = it.nextPartition()
            while (it.hasNext && it.nextPartition() == partitionId) { 
              it.writeNext(writer)
            }
            val segment = writer.commitAndGet()
            lengths(partitionId) = segment.length
          }
        } else { 
          //
          // We must perform merge-sort; get an iterator by partition and write everything directly.
          // [重点] 我们必须执行合并排序；按分区获取迭代器并直接编写所有内容。
          for ((id, elements) <- this.partitionedIterator) { 
            if (elements.hasNext) { 
              for (elem <- elements) { 
                writer.write(elem._1, elem._2)
              }
              val segment = writer.commitAndGet()
              lengths(id) = segment.length
            }
          }
        }
    
        writer.close()
        context.taskMetrics().incMemoryBytesSpilled(memoryBytesSpilled)
        context.taskMetrics().incDiskBytesSpilled(diskBytesSpilled)
        context.taskMetrics().incPeakExecutionMemory(peakMemoryUsedBytes)
    
        lengths
      }

### 3.1.11 xxx ###

### 3.1.12 xxx ###

### 3.1.13 写入索引文件 ###

writeIndexFileAndCommit

### 3.1.14 xxx ###

### 3.1.15 xxx ###

/private/var/folders/37/9746t\_yx10v2g49vkwtzjw\_80000gn/T/blockmgr-9c480511-104d-48ac-b2cd-ac8dacbadfca/0c/shuffle\_0\_0\_0.data

## 3.2. BypassMergeSortShuffleWriter ##

## 3.3. UnsafeShuffleWriter ##

# 四.Shuffle Read框架设计和实现 #

在Shuffle Read阶段，数据操作需要3个功能：跨节点数据获取、聚合和排序。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 10]

Spark为了⽀持所有的情况，设计了⼀个通⽤的Shuffle Read框架，框架的计算顺序为“数据获取→聚合→排序”输出

reduce task不断从各个map task的分区⽂件中获取数据（Fetch records），然后使⽤类似HashMap的结构来对数据进⾏聚合（aggregate），该过程是边获取数据边聚合。聚合完成后，将HashMap中的数据放⼊类似Array的数据结构中按照Key进⾏排序（sort by Key），最后将排序结果输出或者传递给下⼀个操作。如果不需要聚合或者排序，则可以去掉相应的聚合或排序过程。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 11]

1，数据获取分为远程获取和本地获取。本地获取将直接从本地的 BlockManager 取数据， 而对于远程数据，需要走网络。在远程获取过程中，有相关参数可以控制从远程并发获取数据的大小，正在获取数据的请求数，以及单次数据块请求是否放到内存等参数。具体参数包括 spark.reducer.maxSizeInFlight (默认 48M)，spark.reducer.maxReqsInFlight， spark.reducer.maxBlocksInFlightPerAddress 和 spark.maxRemoteBlockSizeFetchToMem。考虑到数据倾斜的场景，如果 Map 阶段有一个 Block 数据特别的大，默认情况由于 spark.maxRemoteBlockSizeFetchToMem 没有做限制，所以在这个阶段需要将需要获取的整个 Block 数据放到 Reduce 端的内存中，这个时候是非常的耗内存的。可以设置 spark.maxRemoteBlockSizeFetchToMem 值，如果超过该阈值，可以落盘，避免这种情况的 OOM。 另外，在获取到数据以后，默认情况下会对获取的数据进行校验（参数 spark.shuffle.detectCorrupt 控制），这个过程也增加了一定的内存消耗。

2，对于需要聚合和排序的情况，这个过程是借助 ExternalAppendOnlyMap 来实现的。整个插入，Spill 以及 Merge 的过程和 Write 阶段差不多。总体上，这块也是比较消耗内存的，但是因为有 Spill 操作，当内存不足时，可以将内存数据刷到磁盘，从而释放内存空间。

参考:  
https://mzl9039.github.io/2018/04/12/spark-%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90-shuffle%E5%8E%9F%E7%90%86.html

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221005/3b5283bdc27748eea507f53a588d5636.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221005/92d6be7978844f64b4e3bd2a16b57b59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221005/3c5cf71053094eac957cc94e86b224c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: /images/20221005/2d2e5dfec5d2482b8162cdb308828ca0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]: /images/20221005/e03ba944b4aa4a2f94b23f3685b0dd5c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 5]: /images/20221005/7fbcfa00c3b844e1a486388c006e39b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 6]: /images/20221005/dd0552c9c1e8464d939f79225d5e3113.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 7]: /images/20221005/53b320e424f74e4f8df9c1a488757f6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 8]: /images/20221005/28808cd52d9a4ef8820d0b615e251077.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 9]: /images/20221005/94e94a4b5bde4129a54c18250d4e2f6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 10]: /images/20221005/d3fed6250eaa4a20abfec9da68b376c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 11]: /images/20221005/3a4a2e2b1cbb4ee58837fe2235b0540d.png