Spark Shuffle過(guò)程分析：Map階段處理流程

作者：時(shí)延軍 2017-06-07 12:25:37

默認(rèn)配置情況下，Spark在Shuffle過(guò)程中會(huì)使用SortShuffleManager來(lái)管理Shuffle過(guò)程中需要的基本組件，以及對(duì)RDD各個(gè)Partition數(shù)據(jù)的計(jì)算。我們可以在Driver和Executor對(duì)應(yīng)的SparkEnv對(duì)象創(chuàng)建過(guò)程中看到對(duì)應(yīng)的配置。

// Let the user specify short names for shuffle managers 
    val shortShuffleMgrNames = Map( 
      "sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName, 
      "tungsten-sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName) 
    val shuffleMgrName = conf.get("spark.shuffle.manager", "sort") 
    val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName) 
    val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)

如果需要修改ShuffleManager實(shí)現(xiàn)，則只需要修改配置項(xiàng)spark.shuffle.manager即可，默認(rèn)支持sort和 tungsten-sort，可以指定自己實(shí)現(xiàn)的ShuffleManager類。

因?yàn)镾huffle過(guò)程中需要將Map結(jié)果數(shù)據(jù)輸出到文件，所以需要通過(guò)注冊(cè)一個(gè)ShuffleHandle來(lái)獲取到一個(gè)ShuffleWriter對(duì)象，通過(guò)它來(lái)控制Map階段記錄數(shù)據(jù)輸出的行為。其中，ShuffleHandle包含了如下基本信息：

shuffleId：標(biāo)識(shí)Shuffle過(guò)程的唯一ID
numMaps：RDD對(duì)應(yīng)的Partitioner指定的Partition的個(gè)數(shù)，也就是ShuffleMapTask輸出的Partition個(gè)數(shù)
dependency：RDD對(duì)應(yīng)的依賴ShuffleDependency

下面我們看下，在SortShuffleManager中是如何注冊(cè)Shuffle的，代碼如下所示：

override def registerShuffle[K, V, C]( 
      shuffleId: Int, 
      numMaps: Int, 
      dependency: ShuffleDependency[K, V, C]): ShuffleHandle = { 
    if (SortShuffleWriter.shouldBypassMergeSort(SparkEnv.get.conf, dependency)) { 
      new BypassMergeSortShuffleHandle[K, V]( 
        shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]]) 
    } else if (SortShuffleManager.canUseSerializedShuffle(dependency)) { 
      new SerializedShuffleHandle[K, V]( 
        shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]]) 
    } else { 
      new BaseShuffleHandle(shuffleId, numMaps, dependency) 
    } 
  }

上面代碼中，對(duì)應(yīng)如下3種ShuffleHandle可以選擇，說(shuō)明如下：

BypassMergeSortShuffleHandle

如果dependency不需要進(jìn)行Map Side Combine，并且RDD對(duì)應(yīng)的ShuffleDependency中的Partitioner設(shè)置的Partition的數(shù)量(這個(gè)不要和parent RDD的Partition個(gè)數(shù)混淆，Partitioner指定了map處理結(jié)果的Partition個(gè)數(shù)，每個(gè)Partition數(shù)據(jù)會(huì)在Shuffle過(guò)程中全部被拉取而拷貝到下游的某個(gè)Executor端)小于等于配置參數(shù)spark.shuffle.sort.bypassMergeThreshold的值，則會(huì)注冊(cè)BypassMergeSortShuffleHandle。默認(rèn)情況下，spark.shuffle.sort.bypassMergeThreshold的取值是200，這種情況下會(huì)直接將對(duì)RDD的 map處理結(jié)果的各個(gè)Partition數(shù)據(jù)寫(xiě)入文件，并***做一個(gè)合并處理。

SerializedShuffleHandle

如果ShuffleDependency中的Serializer，允許對(duì)將要輸出數(shù)據(jù)對(duì)象進(jìn)行排序后，再執(zhí)行序列化寫(xiě)入到文件，則會(huì)選擇創(chuàng)建一個(gè)SerializedShuffleHandle。

BaseShuffleHandle

除了上面兩種ShuffleHandle以后，其他情況都會(huì)創(chuàng)建一個(gè)BaseShuffleHandle對(duì)象，它會(huì)以反序列化的格式處理Shuffle輸出數(shù)據(jù)。

Map階段處理流程分析

Map階段RDD的計(jì)算，對(duì)應(yīng)ShuffleMapTask這個(gè)實(shí)現(xiàn)類，它最終會(huì)在每個(gè)Executor上啟動(dòng)運(yùn)行，每個(gè)ShuffleMapTask處理RDD的一個(gè)Partition的數(shù)據(jù)。這個(gè)過(guò)程的核心處理邏輯，代碼如下所示：

val manager = SparkEnv.get.shuffleManager 
      writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context) 
      writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])

上面代碼中，在調(diào)用rdd的iterator()方法時(shí)，會(huì)根據(jù)RDD實(shí)現(xiàn)類的compute方法指定的處理邏輯對(duì)數(shù)據(jù)進(jìn)行處理，當(dāng)然，如果該P(yáng)artition對(duì)應(yīng)的數(shù)據(jù)已經(jīng)處理過(guò)并存儲(chǔ)在MemoryStore或DiskStore，直接通過(guò)BlockManager獲取到對(duì)應(yīng)的Block數(shù)據(jù)，而無(wú)需每次需要時(shí)重新計(jì)算。然后，write()方法會(huì)將已經(jīng)處理過(guò)的Partition數(shù)據(jù)輸出到磁盤(pán)文件。

在Spark Shuffle過(guò)程中，每個(gè)ShuffleMapTask會(huì)通過(guò)配置的ShuffleManager實(shí)現(xiàn)類對(duì)應(yīng)的ShuffleManager對(duì)象(實(shí)際上是在SparkEnv中創(chuàng)建)，根據(jù)已經(jīng)注冊(cè)的ShuffleHandle，獲取到對(duì)應(yīng)的ShuffleWriter對(duì)象，然后通過(guò)ShuffleWriter對(duì)象將Partition數(shù)據(jù)寫(xiě)入內(nèi)存或文件。所以，接下來(lái)我們可能關(guān)心每一種ShuffleHandle對(duì)應(yīng)的ShuffleWriter的行為，可以看到SortShuffleManager中獲取到ShuffleWriter的實(shí)現(xiàn)代碼，如下所示：

/** Get a writer for a given partition. Called on executors by map tasks. */ 
  override def getWriter[K, V]( 
      handle: ShuffleHandle, 
      mapId: Int, 
      context: TaskContext): ShuffleWriter[K, V] = { 
    numMapsForShuffle.putIfAbsent( 
      handle.shuffleId, handle.asInstanceOf[BaseShuffleHandle[_, _, _]].numMaps) 
    val env = SparkEnv.get 
    handle match { 
      case unsafeShuffleHandle: SerializedShuffleHandle[K @unchecked, V @unchecked] => 
        new UnsafeShuffleWriter( 
          env.blockManager, 
          shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver], 
          context.taskMemoryManager(), 
          unsafeShuffleHandle, 
          mapId, 
          context, 
          env.conf) 
      case bypassMergeSortHandle: BypassMergeSortShuffleHandle[K @unchecked, V @unchecked] => 
        new BypassMergeSortShuffleWriter( 
          env.blockManager, 
          shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver], 
          bypassMergeSortHandle, 
          mapId, 
          context, 
          env.conf) 
      case other: BaseShuffleHandle[K @unchecked, V @unchecked, _] => 
        new SortShuffleWriter(shuffleBlockResolver, other, mapId, context) 
    } 
  }

我們以最簡(jiǎn)單的SortShuffleWriter為例進(jìn)行分析，在SortShuffleManager可以通過(guò)getWriter()方法創(chuàng)建一個(gè)SortShuffleWriter對(duì)象，然后在ShuffleMapTask中調(diào)用SortShuffleWriter對(duì)象的write()方法處理Map輸出的記錄數(shù)據(jù)，write()方法的處理代碼，如下所示：

/** Write a bunch of records to this task's output */ 
  override def write(records: Iterator[Product2[K, V]]): Unit = { 
    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) 
    } 
    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) 
    val blockId = ShuffleBlockId(dep.shuffleId, mapId, IndexShuffleBlockResolver.NOOP_REDUCE_ID) 
    val partitionLengths = sorter.writePartitionedFile(blockId, tmp) 
    shuffleBlockResolver.writeIndexFileAndCommit(dep.shuffleId, mapId, partitionLengths, tmp) 
    mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths) 
  }

從SortShuffleWriter類中的write()方法可以看到，最終調(diào)用了ExeternalSorter的insertAll()方法，實(shí)現(xiàn)了Map端RDD某個(gè)Partition數(shù)據(jù)處理并輸出到內(nèi)存或磁盤(pán)文件，這也是處理Map階段輸出記錄數(shù)據(jù)最核心、最復(fù)雜的過(guò)程。我們將其分為兩個(gè)階段進(jìn)行分析：***階段是，ExeternalSorter的insertAll()方法處理過(guò)程，將記錄數(shù)據(jù)Spill到磁盤(pán)文件;第二階段是，執(zhí)行完insertAll()方法之后的處理邏輯，創(chuàng)建Shuffle Block數(shù)據(jù)文件及其索引文件。

內(nèi)存緩沖寫(xiě)記錄數(shù)據(jù)并Spill到磁盤(pán)文件

查看SortShuffleWriter類的write()方法可以看到，在內(nèi)存中緩存記錄數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)有兩種：一種是Buffer，對(duì)應(yīng)的實(shí)現(xiàn)類PartitionedPairBuffer，設(shè)置mapSideCombine=false時(shí)會(huì)使用該結(jié)構(gòu);另一種是Map，對(duì)應(yīng)的實(shí)現(xiàn)類是PartitionedAppendOnlyMap，設(shè)置mapSideCombine=false時(shí)會(huì)使用該結(jié)構(gòu)。根據(jù)是否指定mapSideCombine選項(xiàng)，分別對(duì)應(yīng)不同的處理流程，我們分別說(shuō)明如下：

設(shè)置mapSideCombine=false時(shí)

這種情況在Map階段不進(jìn)行Combine操作，在內(nèi)存中緩存記錄數(shù)據(jù)會(huì)使用PartitionedPairBuffer這種數(shù)據(jù)結(jié)構(gòu)來(lái)緩存、排序記錄數(shù)據(jù)，它是一個(gè)Append-only Buffer，僅支持向Buffer中追加數(shù)據(jù)鍵值對(duì)記錄，PartitionedPairBuffer的結(jié)構(gòu)如下圖所示：

默認(rèn)情況下，PartitionedPairBuffer初始分配的存儲(chǔ)容量為capacity = initialCapacity = 64，實(shí)際上這個(gè)容量是針對(duì)key的容量，因?yàn)橐鎯?chǔ)的是鍵值對(duì)記錄數(shù)據(jù)，所以實(shí)際存儲(chǔ)鍵值對(duì)的容量為2*initialCapacity = 128。PartitionedPairBuffer是一個(gè)能夠動(dòng)態(tài)擴(kuò)充容量的Buffer，內(nèi)部使用一個(gè)一維數(shù)組來(lái)存儲(chǔ)鍵值對(duì)，每次擴(kuò)容結(jié)果為當(dāng)前Buffer容量的2倍，即2*capacity，***支持存儲(chǔ)2^31-1個(gè)鍵值對(duì)記錄(1073741823個(gè))。

通過(guò)上圖可以看到，PartitionedPairBuffer存儲(chǔ)的鍵值對(duì)記錄數(shù)據(jù)，鍵是(partition, key)這樣一個(gè)Tuple，值是對(duì)應(yīng)的數(shù)據(jù)value，而且curSize是用來(lái)跟蹤寫(xiě)入Buffer中的記錄的，key在Buffer中的索引位置為2*curSize，value的索引位置為2*curSize+1，可見(jiàn)一個(gè)鍵值對(duì)的key和value的存儲(chǔ)在PartitionedPairBuffer內(nèi)部的數(shù)組中是相鄰的。

使用PartitionedPairBuffer緩存鍵值對(duì)記錄數(shù)據(jù)，通過(guò)跟蹤實(shí)際寫(xiě)入到Buffer內(nèi)的記錄數(shù)據(jù)的字節(jié)數(shù)來(lái)判斷，是否需要將Buffer中的數(shù)據(jù)Spill到磁盤(pán)文件，如下代碼所示：

protected def maybeSpill(collection: C, currentMemory: Long): Boolean = { 
    var shouldSpill = false 
    if (elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold) { 
      // 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 
    } 
    shouldSpill = shouldSpill || _elementsRead > numElementsForceSpillThreshold 
    // Actually spill 
    if (shouldSpill) { 
      _spillCount += 1 
      logSpillage(currentMemory) 
      spill(collection) 
      _elementsRead = 0 
      _memoryBytesSpilled += currentMemory 
      releaseMemory() 
    } 
    shouldSpill 
  }

上面elementsRead表示存儲(chǔ)到PartitionedPairBuffer中的記錄數(shù)，currentMemory是對(duì)Buffer中的總記錄數(shù)據(jù)大小(字節(jié)數(shù))的估算，myMemoryThreshold通過(guò)配置項(xiàng)spark.shuffle.spill.initialMemoryThreshold來(lái)進(jìn)行設(shè)置的，默認(rèn)值為5 * 1024 * 1024 = 5M。當(dāng)滿足條件elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold時(shí)，會(huì)先嘗試向MemoryManager申請(qǐng)2 * currentMemory – myMemoryThreshold大小的內(nèi)存，如果能夠申請(qǐng)到，則不進(jìn)行Spill操作，而是繼續(xù)向Buffer中存儲(chǔ)數(shù)據(jù)，否則就會(huì)調(diào)用spill()方法將Buffer中數(shù)據(jù)輸出到磁盤(pán)文件。

向PartitionedPairBuffer中寫(xiě)入記錄數(shù)據(jù)，以及滿足條件Spill記錄數(shù)據(jù)到磁盤(pán)文件，具體處理流程，如下圖所示：

為了查看按照怎樣的規(guī)則進(jìn)行排序，我們看一下，當(dāng)不進(jìn)行Map Side Combine時(shí)，創(chuàng)建ExternalSorter對(duì)象的代碼如下所示：

// 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)

上面aggregator = None，ordering = None，在對(duì)PartitionedPairBuffer中的記錄數(shù)據(jù)Spill到磁盤(pán)之前，要使用默認(rèn)的排序規(guī)則進(jìn)行排序，排序的規(guī)則是只對(duì)PartitionedPairBuffer中的記錄按Partition ID進(jìn)行升序排序，可以查看WritablePartitionedPairCollection伴生對(duì)象類的代碼(其中PartitionedPairBuffer類實(shí)現(xiàn)了特質(zhì)WritablePartitionedPairCollection)，如下所示：

/** 
   * A comparator for (Int, K) pairs that orders them by only their partition ID. 
   */ 
  def partitionComparator[K]: Comparator[(Int, K)] = new Comparator[(Int, K)] { 
    override def compare(a: (Int, K), b: (Int, K)): Int = { 
      a._1 - b._1 
    } 
  }

上面圖中，引用了SortShuffleWriter.writeBlockFiles這個(gè)子序列圖，用來(lái)生成Block數(shù)據(jù)文件和索引文件，后面我們會(huì)單獨(dú)說(shuō)明。通過(guò)對(duì)RDD進(jìn)行計(jì)算生成一個(gè)記錄迭代器對(duì)象，通過(guò)該迭代器迭代出的記錄會(huì)存儲(chǔ)到PartitionedPairBuffer中，當(dāng)滿足Spill條件時(shí)，先對(duì)PartitionedPairBuffer中記錄進(jìn)行排序，***Spill到磁盤(pán)文件，這個(gè)過(guò)程中PartitionedPairBuffer中的記錄數(shù)據(jù)的變化情況，如下圖所示：

上圖中，對(duì)內(nèi)存中PartitionedPairBuffer中的記錄按照Partition ID進(jìn)行排序，并且屬于同一個(gè)Partition的數(shù)據(jù)記錄在PartitionedPairBuffer內(nèi)部的data數(shù)組中是連續(xù)的。排序結(jié)束后，在Spill到磁盤(pán)文件時(shí)，將對(duì)應(yīng)的Partition ID去掉了，只在文件temp_shuffle_4c4b258d-52e4-47a0-a9b6-692f1af7ec9d中連續(xù)存儲(chǔ)鍵值對(duì)數(shù)據(jù)，但同時(shí)在另一個(gè)內(nèi)存數(shù)組結(jié)構(gòu)中會(huì)保存文件中每個(gè)Partition擁有的記錄數(shù)，這樣就能根據(jù)Partition的記錄數(shù)來(lái)順序讀取文件temp_shuffle_4c4b258d-52e4-47a0-a9b6-692f1af7ec9d中屬于同一個(gè)Partition的全部記錄數(shù)據(jù)。

ExternalSorter類內(nèi)部維護(hù)了一個(gè)SpillFile的ArrayBuffer數(shù)組，最終可能會(huì)生成多個(gè)SpillFile，SpillFile的定義如下所示：

private[this] case class SpilledFile( 
    file: File, 
    blockId: BlockId, 
    serializerBatchSizes: Array[Long], 
    elementsPerPartition: Array[Long])

每個(gè)SpillFile包含一個(gè)blockId，標(biāo)識(shí)Map輸出的該臨時(shí)文件;serializerBatchSizes表示每次批量寫(xiě)入到文件的Object的數(shù)量，默認(rèn)為10000，由配置項(xiàng)spark.shuffle.spill.batchSize來(lái)控制;elementsPerPartition表示每個(gè)Partition中的Object的數(shù)量。調(diào)用ExternalSorter的insertAll()方法，最終可能有如下3種情況：

Map階段輸出記錄數(shù)較少，沒(méi)有生成SpillFile，那么所有數(shù)據(jù)都在Buffer中，直接對(duì)Buffer中記錄排序并輸出到文件
Map階段輸出記錄數(shù)較多，生成多個(gè)SpillFile，同時(shí)Buffer中也有部分記錄數(shù)據(jù)
Map階段輸出記錄數(shù)較多，只生成多個(gè)SpillFile
有關(guān)后續(xù)如何對(duì)上面3種情況進(jìn)行處理，可以想見(jiàn)后面對(duì)子序列圖SortShuffleWriter.writeBlockFiles的說(shuō)明。
設(shè)置mapSideCombine=true時(shí)

這種情況在Map階段會(huì)執(zhí)行Combine操作，在Map階段進(jìn)行Combine操作能夠降低Map階段數(shù)據(jù)記錄的總數(shù)，從而降低Shuffle過(guò)程中數(shù)據(jù)的跨網(wǎng)絡(luò)拷貝傳輸。這時(shí)，RDD對(duì)應(yīng)的ShuffleDependency需要設(shè)置一個(gè)Aggregator用來(lái)執(zhí)行Combine操作，可以看下Aggregator類聲明，代碼如下所示：

/** 
 * :: DeveloperApi :: 
 * A set of functions used to aggregate data. 
 * 
 * @param createCombiner function to create the initial value of the aggregation. 
 * @param mergeValue function to merge a new value into the aggregation result. 
 * @param mergeCombiners function to merge outputs from multiple mergeValue function. 
 */ 
@DeveloperApi 
case class Aggregator[K, V, C] ( 
    createCombiner: V => C, 
    mergeValue: (C, V) => C, 
    mergeCombiners: (C, C) => C) { 
  ... ... 
}

由于在Map階段只用到了構(gòu)造Aggregator的幾個(gè)函數(shù)參數(shù)createCombiner、mergeValue、mergeCombiners，我們對(duì)這幾個(gè)函數(shù)詳細(xì)說(shuō)明如下：

createCombiner：進(jìn)行Aggregation開(kāi)始時(shí)，需要設(shè)置初始值。因?yàn)樵贏ggregation過(guò)程中使用了類似Map的內(nèi)存數(shù)據(jù)結(jié)構(gòu)來(lái)管理鍵值對(duì)，每次加入前會(huì)先查看Map內(nèi)存結(jié)構(gòu)中是否存在Key對(duì)應(yīng)的Value，***次肯定不存在，所以***將某個(gè)Key的Value加入到Map內(nèi)存結(jié)構(gòu)中時(shí)，Key在Map內(nèi)存結(jié)構(gòu)中***次有了Value。
mergeValue：某個(gè)Key已經(jīng)在Map結(jié)構(gòu)中存在Value，后續(xù)某次又遇到相同的Key和一個(gè)新的Value，這時(shí)需要通過(guò)該函數(shù)，將舊Value和新Value進(jìn)行合并，根據(jù)Key檢索能夠得到合并后的新Value。
mergeCombiners：一個(gè)Map內(nèi)存結(jié)構(gòu)中Key和Value是由mergeValue生成的，那么在向Map中插入數(shù)據(jù)，肯定會(huì)遇到Map使用容量達(dá)到上限，這時(shí)需要將記錄數(shù)據(jù)Spill到磁盤(pán)文件，那么多個(gè)Spill輸出的磁盤(pán)文件中可能存在同一個(gè)Key，這時(shí)需要對(duì)多個(gè)Spill輸出的磁盤(pán)文件中的Key的多個(gè)Value進(jìn)行合并，這時(shí)需要使用mergeCombiners函數(shù)進(jìn)行處理。

該類中定義了combineValuesByKey、combineValuesByKey、combineCombinersByKey，由于這些函數(shù)是在Reduce階段使用的，所以在這里先不說(shuō)明，后續(xù)文章我們會(huì)單獨(dú)詳細(xì)來(lái)分析。

我們通過(guò)下面的序列圖來(lái)描述，需要進(jìn)行Map Side Combine時(shí)的處理流程，如下所示：

對(duì)照上圖，我們看一下，當(dāng)需要進(jìn)行Map Side Combine時(shí)，對(duì)應(yīng)的ExternalSorter類insertAll()方法中的處理邏輯，代碼如下所示：

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.changeValue((getPartition(kv._1), kv._1), update) 
        maybeSpillCollection(usingMap = true) 
      } 
    }

上面代碼中，map是內(nèi)存數(shù)據(jù)結(jié)構(gòu)，最重要的是update函數(shù)和map的changeValue方法(這里的map對(duì)應(yīng)的實(shí)現(xiàn)類是PartitionedAppendOnlyMap)。update函數(shù)所做的工作，其實(shí)就是對(duì)createCombiner和mergeValue這兩個(gè)函數(shù)的使用，***次遇到一個(gè)Key調(diào)用createCombiner函數(shù)處理，非***遇到同一個(gè)Key對(duì)應(yīng)新的Value調(diào)用mergeValue函數(shù)進(jìn)行合并處理。map的changeValue方法主要是將Key和Value在map中存儲(chǔ)或者進(jìn)行修改(對(duì)出現(xiàn)的同一個(gè)Key的多個(gè)Value進(jìn)行合并，并將合并后的新Value替換舊Value)。

PartitionedAppendOnlyMap是一個(gè)經(jīng)過(guò)優(yōu)化的哈希表，它支持向map中追加數(shù)據(jù)，以及修改Key對(duì)應(yīng)的Value，但是不支持刪除某個(gè)Key及其對(duì)應(yīng)的Value。它能夠支持的存儲(chǔ)容量是0.7 * 2 ^ 29 = 375809638。當(dāng)達(dá)到指定存儲(chǔ)容量或者指定限制，就會(huì)將map中記錄數(shù)據(jù)Spill到磁盤(pán)文件，這個(gè)過(guò)程和前面的類似，不再累述。

創(chuàng)建Shuffle Block數(shù)據(jù)文件及其索引文件

無(wú)論是使用PartitionedPairBuffer，還是使用PartitionedAppendOnlyMap，當(dāng)需要容量滿足Spill條件時(shí)，都會(huì)將該內(nèi)存結(jié)構(gòu)(buffer/map)中記錄數(shù)據(jù)Spill到磁盤(pán)文件，所以Spill到磁盤(pán)文件的格式是相同的。對(duì)于后續(xù)Block數(shù)據(jù)文件和索引文件的生成邏輯也是相同，如下圖所示：

假設(shè)，我們生成的Shuffle Block文件對(duì)應(yīng)各個(gè)參數(shù)為：shuffleId=2901，mapId=11825，reduceId=0，這里reduceId是一個(gè)NOOP_REDUCE_ID，表示與DiskStore進(jìn)行磁盤(pán)I/O交互操作，而DiskStore期望對(duì)應(yīng)一個(gè)(map, reduce)對(duì)，但是對(duì)于排序的Shuffle輸出，通常Reducer拉取數(shù)據(jù)后只生成一個(gè)文件(Reduce文件)，所以這里默認(rèn)reduceId為0。經(jīng)過(guò)上圖的處理流程，可以生成一個(gè).data文件，也就是Block數(shù)據(jù)文件;一個(gè).index文件，也就是包含了各個(gè)Partition在數(shù)據(jù)文件中的偏移位置的索引文件。這個(gè)過(guò)程生成的文件，示例如下所示：

shuffle_2901_11825_0.data  
shuffle_2901_11825_0.index

這樣，對(duì)于每個(gè)RDD的多個(gè)Partition進(jìn)行處理后，都會(huì)生成對(duì)應(yīng)的數(shù)據(jù)文件和索引文件，后續(xù)在Reduce端就可以讀取這些Block文件，這些記錄數(shù)據(jù)在文件中都是經(jīng)過(guò)分區(qū)(Partitioned)的。

責(zé)任編輯：未麗燕來(lái)源： 36大數(shù)據(jù)