1.  /**
      *使用Hash的ShuffleManager具體實現子類，針對每個Mapper都會為各個Reduce分
      *區構建一個輸出文件（也可能是多個任務復用文件）
2.    */
3.  private[spark]      class    HashShuffleManager(conf:         SparkConf)   extends
    ShuffleManager with Logging {
4.  ......

1.  private[spark] class HashShuffleWriter[K, V](
2.      shuffleBlockResolver: FileShuffleBlockResolver,
3.      handle: BaseShuffleHandle[K, V, _],
4.  mapId: Int,
5.      context: TaskContext)
6.    extends ShuffleWriter[K, V] with Logging {
7.
8.    //控制每個Writer輸出時的切片個數，對應分區個數
9.    private val dep = handle.dependency
10.   private val numOutputSplits = dep.partitioner.numPartitions
11.
12.   ......
13.   //獲取數據讀寫的塊管理器
14.   private val blockManager = SparkEnv.get.blockManager
15.   private val ser = Serializer.getSerializer(dep.serializer.getOrElse(null))
16.
17. //從  FileShuffleBlockResolver      的 forMapTask方法中獲取指定的    shuffleId  對應
    //的mapId
18. //對應分區個數構建的數據塊寫的ShuffleWriterGroup實例
19.   private val shuffle = shuffleBlockResolver.forMapTask(dep.shuffleId,
      mapId, numOutputSplits, ser, writeMetrics)
20.
21.
22.   /** Task輸出時一組記錄的寫入 */
23.
24.   override def write(records: Iterator[Product2[K, V]]): Unit = {
25. //判斷在寫時是否需要先聚合，即定義了Map端Combine時，先對數據進行聚合再寫入，否則
    //直接返回需要寫入的一批記錄
26.
27.   val iter = if (dep.aggregator.isDefined) {
28.       if (dep.mapSideCombine) {
29.         dep.aggregator.get.combineValuesByKey(records, context)
30.       } else {
31.         records
32.       }
33.     } else {
34.       require(!dep.mapSideCombine, "Map-side combine without Aggregator
          specified!")
35.       records
36.     }
37.
38.     //根據分區器，獲取每條記錄對應的bucketId（即所在Reduce序號），根據bucketId
        //從FileShuffleBlockResolver構建的ShuffleWriterGroup中，獲取DiskBlock-
        //ObjectWriter實例，對應磁盤數據塊的數據寫入器
39.     for (elem<- iter) {
40.       val bucketId = dep.partitioner.getPartition(elem._1)
41.       shuffle.writers(bucketId).write(elem._1, elem._2)
42.     }
43.   }
44.      ......
45. }

1.  /**
2.    *針對給定的 Map Task，指定一個ShuffleWriterGroup實例，在數據塊寫入器成功
3.    *關閉時，會注冊為完成狀態
4.    */
5.
6.
7.   def forMapTask(shuffleId: Int, mapId: Int, numReduces: Int, serializer:
     Serializer,
8.  writeMetrics: ShuffleWriteMetrics): ShuffleWriterGroup = {
9.      new ShuffleWriterGroup {
10.       //在FileShuffleBlockResolver中維護著當前Map Task對應shuffleId標識的
          //Shuffle中，指定numReduces個數的Reduce的各個狀態
11.
12. shuffleStates.putIfAbsent(shuffleId, new ShuffleState(numReduces))
13.       private val shuffleState = shuffleStates(shuffleId)
14.
15. ......
16.           //根據Reduce端的任務個數，構建元素類型為DiskBlockObjectWriter的數組，
              //DiskBlockObjectWriter負責具體數據的磁盤寫入
17.    //原則上，Shuffle的輸出可以存放在各種提供存儲機制的系統上，但為了容錯性等方面的
       //考慮，目前的Shuffle實行機制都會寫入到磁盤中
18.
19.  val writers: Array[DiskBlockObjectWriter] = {
20.         //這里的邏輯Bucket的Id值即對應的Reduce的任務序號，或者說分區ID
21.         Array.tabulate[DiskBlockObjectWriter](numReduces) { bucketId =>
22.         //針對每個Map端分區的Id與Bucket的Id構建數據塊的邏輯標識
23.           val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
24.           val blockFile = blockManager.diskBlockManager.getFile(blockId)
25.           val tmp = Utils.tempFileWith(blockFile)
26.blockManager.getDiskWriter(blockId, tmp, serializerInstance, bufferSize,
    writeMetrics)
27.         }
28.       }
29. ......
30.       //任務完成時回調的釋放寫入器方法
31.       override def releaseWriters(success: Boolean) {
32. shuffleState.completedMapTasks.add(mapId)
33.       }
34.     }
35.   }

1.  case class ShuffleBlockId(shuffleId: Int, mapId: Int, reduceId: Int)
    extends BlockId {
2.    override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_"
      + reduceId
3.  }

1.  /**
2.    *獲取一個針對特定Map Task的ShuffleWriterGroup
3.    */
4.
5.
6.    def forMapTask(shuffleId: Int, mapId: Int, numBuckets: Int, serializer:
      Serializer,
7.  writeMetrics: ShuffleWriteMetrics): ShuffleWriterGroup = {
8.      new ShuffleWriterGroup {
9.  ......
10.       val writers: Array[DiskBlockObjectWriter] = if
         (consolidateShuffleFiles) {
11. //獲取未使用的文件組
12. fileGroup = getUnusedFileGroup()
13.         Array.tabulate[DiskBlockObjectWriter](numBuckets) { bucketId =>
14. val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
15. //注意獲取磁盤寫入器時，傳入的第二個參數與未使用文件合并機制時的差異
16. //fileGroup(bucketId)：構造器方式調用，對應apply的方法調用
17.blockManager.getDiskWriter(blockId, fileGroup(bucketId), serializerInstance,
    bufferSize,
18. writeMetrics)
19.         }
20.       } else {
21.         Array.tabulate[DiskBlockObjectWriter](numBuckets) { bucketId =>
22. val blockId = ShuffleBlockId(shuffleId, mapId, bucketId)
23. //根據ShuffleBlockId信息獲取文件名
24.           val blockFile = blockManager.diskBlockManager.getFile(blockId)
25.           val tmp = Utils.tempFileWith(blockFile)
26.blockManager.getDiskWriter(blockId, tmp, serializerInstance, bufferSize,
    writeMetrics)
27.         }
28.       }
29. ......
30. writeMetrics.incShuffleWriteTime(System.nanoTime - openStartTime)
31.       override def releaseWriters(success: Boolean) {
32. //帶文件合并機制時，寫入器在釋放后的處理
33. //3個關鍵信息mapId、offsets、lengths
34.         if (consolidateShuffleFiles) {
35.           if (success) {
36.             val offsets = writers.map(_.fileSegment().offset)
37.             val lengths = writers.map(_.fileSegment().length)
38. fileGroup.recordMapOutput(mapId, offsets, lengths)
39.           }
40. //回收文件組，便于后續復用
41. recycleFileGroup(fileGroup)
42.         } else {
43. shuffleState.completedMapTasks.add(mapId)
44.         }
45.       }

1.  private def getUnusedFileGroup(): ShuffleFileGroup = {
2.  //獲取已經構建但未使用的文件組，如果獲取失敗，則重新構建一個文件組
3.          val fileGroup = shuffleState.unusedFileGroups.poll()
4.          if (fileGroup != null) fileGroup else newFileGroup()
5.        }
6.  //重新構建一個文件組的源碼
7.        private def newFileGroup(): ShuffleFileGroup = {
8.  //構建后會對文件編號進行遞增，該文件編號最終用在生成的文件名中
9.          val fileId = shuffleState.nextFileId.getAndIncrement()
10.         val files = Array.tabulate[File](numBuckets) { bucketId =>
11. //最終的文件名，可以通過文件名的組成及取值細節，加深對實現細節在文件個數上的差異的理解
12.
13.           val filename = physicalFileName(shuffleId, bucketId, fileId)
14. blockManager.diskBlockManager.getFile(filename)
15.         }
16. //構建并添加到shuffleState中，便于后續復用
17.         val fileGroup = new ShuffleFileGroup(shuffleId, fileId, files)
18. shuffleState.allFileGroups.add(fileGroup)
19. fileGroup
20.       }

1.  private def physicalFileName(shuffleId: Int, bucketId: Int, fileId: Int) = {
2.      "merged_shuffle_%d_%d_%d".format(shuffleId, bucketId, fileId)
3.    }

1.  private def resourceOfferSingleTaskSet(
2.  taskSet: TaskSetManager,
3.  maxLocality: TaskLocality,
4.  shuffledOffers: Seq[WorkerOffer],
5.  availableCpus: Array[Int],
6.      tasks: Seq[ArrayBuffer[TaskDescription]]) : Boolean = {
7.  var launchedTask = false
8.    for (i <- 0 until shuffledOffers.size) {
9.  val execId = shuffledOffers(i).executorId
10. val host = shuffledOffers(i).host
11. //判斷當前Executor工作點上可用的內核個數是否滿足Task所需的內核個數
12. //CPUS_PER_TASK：表示設置的每個Task所需的內核個數
13. if (availableCpus(i) >= CPUS_PER_TASK) {
14. try {
15. for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
16. ......
17. launchedTask = true
18. }
19.      } catch {
20. ......
21.       }
22.     }
23.   }
24. return launchedTask
25. }

1.  //每個任務請求的CPU個數
2.  val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1)