// DAGScheduler.scala
private def submitMissingTasks(stage: Stage, jobId: Int) {logDebug("submitMissingTasks(" + stage + ")")// First figure out the indexes of partition ids to compute.val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()...// ResultStage.scala
override def findMissingPartitions(): Seq[Int] = {val job = activeJob.get(0 until job.numPartitions).filter(id => !job.finished(id))
}
// ActiveJob.scala
val numPartitions = finalStage match {// 对于ResultStage，不一定得到当前rdd的所有分区，例如first()和lookup()的Action，// 因此这里是r.partitions而不是r.rdd.partitionscase r: ResultStage => r.partitions.lengthcase m: ShuffleMapStage => m.rdd.partitions.length
}// ShuffleMapStage.scala
override def findMissingPartitions(): Seq[Int] = {mapOutputTrackerMaster.findMissingPartitions(shuffleDep.shuffleId).getOrElse(0 until numPartitions)
}// MapOutputTrackerMaster.scala
def findMissingPartitions(shuffleId: Int): Option[Seq[Int]] = {shuffleStatuses.get(shuffleId).map(_.findMissingPartitions())
}

// DAGScheduler.scala...// Use the scheduling pool, job group, description, etc. from an ActiveJob associated// with this Stageval properties = jobIdToActiveJob(jobId).propertiesrunningStages += stage// 在检测Tasks是否serializable之前，就要SparkListenerStageSubmitted，// 如果不能serializable，那就在这**之后**给一个SparkListenerStageCompletedstage match {case s: ShuffleMapStage =>outputCommitCoordinator.stageStart(stage = s.id, maxPartitionId = s.numPartitions - 1)case s: ResultStage =>outputCommitCoordinator.stageStart(stage = s.id, maxPartitionId = s.rdd.partitions.length - 1)}...

// DAGScheduler.scala...val taskIdToLocations: Map[Int, Seq[TaskLocation]] = try {stage match {case s: ShuffleMapStage =>partitionsToCompute.map { id => (id, getPreferredLocs(stage.rdd, id))}.toMapcase s: ResultStage =>partitionsToCompute.map { id =>val p = s.partitions(id)(id, getPreferredLocs(stage.rdd, p))}.toMap}} catch {case NonFatal(e) =>// 如果有非致命异常就创建一个新的Attempt，并且abortStage（这还不致命么）stage.makeNewStageAttempt(partitionsToCompute.size)listenerBus.post(SparkListenerStageSubmitted(stage.latestInfo, properties))abortStage(stage, s"Task creation failed: $en${Utils.exceptionString(e)}", Some(e))runningStages -= stagereturn}...

// DAGScheduler.scala...stage.makeNewStageAttempt(partitionsToCompute.size, taskIdToLocations.values.toSeq)// 如果没有Task要执行，实际上就是skip了，那么就没有Submission Time这个字段if (partitionsToCompute.nonEmpty) {stage.latestInfo.submissionTime = Some(clock.getTimeMillis())}listenerBus.post(SparkListenerStageSubmitted(stage.latestInfo, properties))// TODO: 也许可以将`taskBinary`放到Stage里面以避免对它序列化多次。// 一堆注释看不懂var taskBinary: Broadcast[Array[Byte]] = nullvar partitions: Array[Partition] = nulltry {var taskBinaryBytes: Array[Byte] = null// taskBinaryBytes and partitions are both effected by the checkpoint status. We need// this synchronization in case another concurrent job is checkpointing this RDD, so we get a// consistent view of both variables.RDDCheckpointData.synchronized {taskBinaryBytes = stage match {case stage: ShuffleMapStage =>JavaUtils.bufferToArray(closureSerializer.serialize((stage.rdd, stage.shuffleDep): AnyRef))case stage: ResultStage =>// 注意这里的stage.func已经被ClosureCleaner清理过了JavaUtils.bufferToArray(closureSerializer.serialize((stage.rdd, stage.func): AnyRef))}partitions = stage.rdd.partitions}...// 广播taskBinary = sc.broadcast(taskBinaryBytes)} catch {// In the case of a failure during serialization, abort the stage.case e: NotSerializableException =>abortStage(stage, "Task not serializable: " + e.toString, Some(e))runningStages -= stage...}

// DAGScheduler.scala...val tasks: Seq[Task[_]] = try {val serializedTaskMetrics = closureSerializer.serialize(stage.latestInfo.taskMetrics).array()stage match {case stage: ShuffleMapStage =>stage.pendingPartitions.clear()partitionsToCompute.map { id =>val locs = taskIdToLocations(id)val part = partitions(id)stage.pendingPartitions += idnew ShuffleMapTask(stage.id, stage.latestInfo.attemptNumber,taskBinary, part, locs, properties, serializedTaskMetrics, Option(jobId),Option(sc.applicationId), sc.applicationAttemptId, stage.rdd.isBarrier())}case stage: ResultStage =>partitionsToCompute.map { id =>val p: Int = stage.partitions(id)val part = partitions(p)val locs = taskIdToLocations(id)new ResultTask(stage.id, stage.latestInfo.attemptNumber,taskBinary, part, locs, id, properties, serializedTaskMetrics,Option(jobId), Option(sc.applicationId), sc.applicationAttemptId,stage.rdd.isBarrier())}}} catch {...}

// DAGScheduler.scala...if (tasks.nonEmpty) {logInfo(s"Submitting ${tasks.size} missing tasks from $stage (${stage.rdd}) (first 15 " +s"tasks are for partitions ${tasks.take(15).map(_.partitionId)})")taskScheduler.submitTasks(new TaskSet(tasks.toArray, stage.id, stage.latestInfo.attemptNumber, jobId, properties))} else {

// Because we posted SparkListenerStageSubmitted earlier, we should mark// the stage as completed here in case there are no tasks to runmarkStageAsFinished(stage, None)stage match {case stage: ShuffleMapStage =>logDebug(s"Stage ${stage} is actually done; " +s"(available: ${stage.isAvailable}," +s"available outputs: ${stage.numAvailableOutputs}," +s"partitions: ${stage.numPartitions})")markMapStageJobsAsFinished(stage)case stage : ResultStage =>logDebug(s"Stage ${stage} is actually done; (partitions: ${stage.numPartitions})")}submitWaitingChildStages(stage)}
}

override def compute(split: Partition, context: TaskContext): Iterator[(K, C)] = {val dep = dependencies.head.asInstanceOf[ShuffleDependency[K, V, C]]val metrics = context.taskMetrics().createTempShuffleReadMetrics()SparkEnv.get.shuffleManager // 由SparkEnv维护的ShuffleManager
...

....getReader(dep.shuffleHandle, split.index, split.index + 1, context, metrics) // 返回一个BlockStoreShuffleReader.read().asInstanceOf[Iterator[(K, C)]]
}

// BlockStoreShuffleReader.scala
override def read(): Iterator[Product2[K, C]] = {val wrappedStreams = new ShuffleBlockFetcherIterator(...) // 返回一个ShuffleBlockFetcherIterator.toCompletionIterator // 返回一个Iterator[(BlockId, InputStream)]

val serializerInstance = dep.serializer.newInstance()// Create a key/value iterator for each streamval recordIter = wrappedStreams.flatMap { case (blockId, wrappedStream) =>serializerInstance.deserializeStream(wrappedStream).asKeyValueIterator}// Update the context task metrics for each record read.// CompletionIterator相比普通的Iterator的区别就是在结束之后会调用一个completion函数// CompletionIterator通过它伴生对象的apply方法创建，传入第二个参数即completionFunctionval metricIter = CompletionIterator[(Any, Any), Iterator[(Any, Any)]](recordIter.map { record =>readMetrics.incRecordsRead(1)record},context.taskMetrics().mergeShuffleReadMetrics())// An interruptible iterator must be used here in order to support task cancellationval interruptibleIter = new InterruptibleIterator[(Any, Any)](context, metricIter "(Any, Any)")...

// BlockStoreShuffleReader.scala...val aggregatedIter: Iterator[Product2[K, C]] = if (dep.aggregator.isDefined) {if (dep.mapSideCombine) {// We are reading values that are already combinedval combinedKeyValuesIterator = interruptibleIter.asInstanceOf[Iterator[(K, C)]]dep.aggregator.get.combineCombinersByKey(combinedKeyValuesIterator, context)} else {// We don't know the value type, but also don't care -- the dependency *should*// have made sure its compatible w/ this aggregator, which will convert the value// type to the combined type Cval keyValuesIterator = interruptibleIter.asInstanceOf[Iterator[(K, Nothing)]]dep.aggregator.get.combineValuesByKey(keyValuesIterator, context)}} else {interruptibleIter.asInstanceOf[Iterator[Product2[K, C]]]}

// Aggregator.scala
case class Aggregator[K, V, C] (createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C) {def combineValuesByKey(iter: Iterator[_ <: Product2[K, V]],context: TaskContext): Iterator[(K, C)] = {val combiners = new ExternalAppendOnlyMap[K, V, C](createCombiner, mergeValue, mergeCombiners "K, V, C")combiners.insertAll(iter)updateMetrics(context, combiners)combiners.iterator}def combineCombinersByKey(iter: Iterator[_ <: Product2[K, C]],context: TaskContext): Iterator[(K, C)] = {val combiners = new ExternalAppendOnlyMap[K, C, C](identity, mergeCombiners, mergeCombiners "K, C, C")// 同上}/** Update task metrics after populating the external map. */private def updateMetrics(context: TaskContext, map: ExternalAppendOnlyMap[_, _, _]): Unit = {Option(context).foreach { c =>c.taskMetrics().incMemoryBytesSpilled(map.memoryBytesSpilled)c.taskMetrics().incDiskBytesSpilled(map.diskBytesSpilled)c.taskMetrics().incPeakExecutionMemory(map.peakMemoryUsedBytes)}}
}

// BlockStoreShuffleReader.scala
...val resultIter = dep.keyOrdering match {case Some(keyOrd: Ordering[K]) =>val sorter = new ExternalSorter[K, C, C](context, ordering = Some(keyOrd "K, C, C"), serializer = dep.serializer)sorter.insertAll(aggregatedIter)context.taskMetrics().incMemoryBytesSpilled(sorter.memoryBytesSpilled)context.taskMetrics().incDiskBytesSpilled(sorter.diskBytesSpilled)context.taskMetrics().incPeakExecutionMemory(sorter.peakMemoryUsedBytes)// Use completion callback to stop sorter if task was finished/cancelled.context.addTaskCompletionListener[Unit](_ => {sorter.stop()})CompletionIterator[Product2[K, C], Iterator[Product2[K, C]]](sorter.iterator, sorter.stop( "Product2[K, C], Iterator[Product2[K, C]]"))case None =>aggregatedIter}

@volatile private[collection] var currentMap = new SizeTrackingAppendOnlyMap[K, C]
private val spilledMaps = new ArrayBuffer[DiskMapIterator]

// AppendOnlyMap.scala
def insertAll(entries: Iterator[Product2[K, V]]): Unit = {if (currentMap == null) {throw new IllegalStateException("Cannot insert new elements into a map after calling iterator")}// 我们复用update函数，从而避免每一次都创建一个新的闭包（编程环境这么恶劣的么。。。）var curEntry: Product2[K, V] = nullval update: (Boolean, C) => C = (hadVal, oldVal) => {if (hadVal)// 如果不是第一个V，就merge// mergeValue: (C, V) => C,mergeValue(oldVal, curEntry._2)else// 如果是第一个V，就新建一个C// createCombiner: V => C,createCombiner(curEntry._2)}while (entries.hasNext) {curEntry = entries.next()val estimatedSize = currentMap.estimateSize()if (estimatedSize > _peakMemoryUsedBytes) {_peakMemoryUsedBytes = estimatedSize}if (maybeSpill(currentMap, estimatedSize)) {// 如果发生了Spill，就重新创建一个currentMapcurrentMap = new SizeTrackingAppendOnlyMap[K, C]}// key: K, updateFunc: (Boolean, C) => CcurrentMap.changeValue(curEntry._1, update)addElementsRead()}
}

// AppendOnlyMap.scala// 这里的nullValue和haveNullValue是用来单独处理k为null的情况的，下面会详细说明private var haveNullValue = false// 有关null.asInstanceOf[V]的花里胡哨的语法，详见 https://stackoverflow.com/questions/10749010/if-an-int-cant-be-null-what-does-null-asinstanceofint-meanprivate var nullValue: V = null.asInstanceOf[V]def changeValue(key: K, updateFunc: (Boolean, V) => V): V = {// updateFunc就是从insertAll传入的updateassert(!destroyed, destructionMessage)val k = key.asInstanceOf[AnyRef]if (k.eq(null)) {if (!haveNullValue) {// 如果这时候还没有null的这个key，就新创建一个incrementSize()}nullValue = updateFunc(haveNullValue, nullValue)haveNullValue = truereturn nullValue}var pos = rehash(k.hashCode) & maskvar i = 1while (true) {// 乘以2的原因是他按照K1 V1 K2 V2这样放的val curKey = data(2 * pos)if (curKey.eq(null)) {// 如果对应的key不存在，就新创建一个// 这也是为什么前面要单独处理null的原因，这里的null被用来做placeholder了// 可以看到，第一个参数传的false，第二个是花里胡哨的nullval newValue = updateFunc(false, null.asInstanceOf[V])data(2 * pos) = kdata(2 * pos + 1) = newValue.asInstanceOf[AnyRef]incrementSize()return newValue} else if (k.eq(curKey) || k.equals(curKey)) { // 又是从Java继承下来的花里胡哨的特性val newValue = updateFunc(true, data(2 * pos + 1).asInstanceOf[V])data(2 * pos + 1) = newValue.asInstanceOf[AnyRef]return newValue} else {// 再散列val delta = ipos = (pos + delta) & maski += 1}}null.asInstanceOf[V] // Never reached but needed to keep compiler happy}

// Spillable.scala
protected def maybeSpill(collection: C, currentMemory: Long): Boolean = {var shouldSpill = falseif (elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold) {val amountToRequest = 2 * currentMemory - myMemoryThreshold// 调用对应MemoryConsumer的acquireMemory方法val granted = acquireMemory(amountToRequest)myMemoryThreshold += grantedshouldSpill = currentMemory >= myMemoryThreshold}shouldSpill = shouldSpill || _elementsRead > numElementsForceSpillThreshold...// MemoryConsumer.scala
public long acquireMemory(long size) {long granted = taskMemoryManager.acquireExecutionMemory(size, this);used += granted;return granted;
}

// Spillable.scala...// Actually spillif (shouldSpill) {_spillCount += 1 // 统计Spill的次数logSpillage(currentMemory)spill(collection)_elementsRead = 0 // 重置强制Spill计数器_elementsRead_memoryBytesSpilled += currentMemoryreleaseMemory()}shouldSpill
}

// ExternalAppendOnlyMap.scala
override def iterator: Iterator[(K, C)] = {...if (spilledMaps.isEmpty) {// 如果没有发生SpilldestructiveIterator(currentMap.iterator)} else {// 如果发生了Spillnew ExternalIterator()}
}def destructiveIterator(inMemoryIterator: Iterator[(K, C)]): Iterator[(K, C)] = {readingIterator = new SpillableIterator(inMemoryIterator)readingIterator.toCompletionIterator
}

// AppendOnlyMap.scala
def nextValue(): (K, V) = {if (pos == -1) {    // Treat position -1 as looking at the null valueif (haveNullValue) {return (null.asInstanceOf[K], nullValue)}pos += 1}while (pos < capacity) {if (!data(2 * pos).eq(null)) {return (data(2 * pos).asInstanceOf[K], data(2 * pos + 1).asInstanceOf[V])}pos += 1}null
}

private[spark] class ExternalSorter[K, V, C](context: TaskContext,aggregator: Option[Aggregator[K, V, C]] = None,partitioner: Option[Partitioner] = None,ordering: Option[Ordering[K]] = None,serializer: Serializer = SparkEnv.get.serializer)extends Spillable[WritablePartitionedPairCollection[K, C]](context.taskMemoryManager( "WritablePartitionedPairCollection[K, C]"))

// ExternalAppendOnlyMap.scala
val mergeHeap = new mutable.PriorityQueue[StreamBuffer]
val sortedMap = destructiveIterator(currentMap.destructiveSortedIterator(keyComparator))
// 我们得到一个Array的迭代器
val inputStreams = (Seq(sortedMap) ++ spilledMaps).map(it => it.buffered)inputStreams.foreach { it =>val kcPairs = new ArrayBuffer[(K, C)]// 读完所有具有所有相同hash(key)的序列，并创建一个StreamBuffer// 需要注意的是，由于哈希碰撞的原因，里面可能有多个keyreadNextHashCode(it, kcPairs)if (kcPairs.length > 0) {mergeHeap.enqueue(new StreamBuffer(it, kcPairs))}
}

// AppendOnlyMap.scala
def destructiveSortedIterator(keyComparator: Comparator[K]): Iterator[(K, V)] = {destroyed = truevar keyIndex, newIndex = 0// 下面这个循环将哈希表里面散乱的KV对压缩到最前面while (keyIndex < capacity) {if (data(2 * keyIndex) != null) {data(2 * newIndex) = data(2 * keyIndex)data(2 * newIndex + 1) = data(2 * keyIndex + 1)newIndex += 1}keyIndex += 1}assert(curSize == newIndex + (if (haveNullValue) 1 else 0))new Sorter(new KVArraySortDataFormat[K, AnyRef]).sort(data, 0, newIndex, keyComparator)// 这下面和前面实现大差不差，就省略了new Iterator[(K, V)] {...}
}

override def next(): (K, C) = {if (mergeHeap.isEmpty) {// 如果堆是空的，就再见了throw new NoSuchElementException}// Select a key from the StreamBuffer that holds the lowest key hash// mergeHeap选择所有StreamBuffer中最小hash的，作为minBufferval minBuffer = mergeHeap.dequeue()// minPairs是一个ArrayBuffer[T]，表示这个StreamBuffer维护的所有KV对val minPairs = minBuffer.pairsval minHash = minBuffer.minKeyHash// 从一个ArrayBuffer[T]中移出Index为0的项目val minPair = removeFromBuffer(minPairs, 0)// 得到非哈希的 (minKey, minCombiner)val minKey = minPair._1var minCombiner = minPair._2assert(hashKey(minPair) == minHash)// For all other streams that may have this key (i.e. have the same minimum key hash),// merge in the corresponding value (if any) from that streamval mergedBuffers = ArrayBuffer[StreamBuffer](minBuffer "StreamBuffer")while (mergeHeap.nonEmpty && mergeHeap.head.minKeyHash == minHash) {val newBuffer = mergeHeap.dequeue()// 如果newBuffer的key和minKey相等的话（考虑哈希碰撞），就合并minCombiner = mergeIfKeyExists(minKey, minCombiner, newBuffer)mergedBuffers += newBuffer}// Repopulate each visited stream buffer and add it back to the queue if it is non-emptymergedBuffers.foreach { buffer =>if (buffer.isEmpty) {readNextHashCode(buffer.iterator, buffer.pairs)}if (!buffer.isEmpty) {mergeHeap.enqueue(buffer)}}(minKey, minCombiner)
}

// SizeTracker.scala
protected def afterUpdate(): Unit = {numUpdates += 1if (nextSampleNum == numUpdates) {takeSample()}
}

// SizeTracker.scala
private def takeSample(): Unit = {samples.enqueue(Sample(SizeEstimator.estimate(this), numUpdates))...

private def estimate(obj: AnyRef, visited: IdentityHashMap[AnyRef, AnyRef]): Long = {val state = new SearchState(visited)state.enqueue(obj)while (!state.isFinished) {visitSingleObject(state.dequeue(), state)}state.size
}

private def visitSingleObject(obj: AnyRef, state: SearchState): Unit = {val cls = obj.getClassif (cls.isArray) {visitArray(obj, cls, state)} else if (cls.getName.startsWith("scala.reflect")) {} else if (obj.isInstanceOf[ClassLoader] || obj.isInstanceOf[Class[_]]) {// Hadoop JobConfs created in the interpreter have a ClassLoader.} else {obj match {case s: KnownSizeEstimation =>state.size += s.estimatedSizecase _ =>val classInfo = getClassInfo(cls)state.size += alignSize(classInfo.shellSize)for (field <- classInfo.pointerFields) {state.enqueue(field.get(obj))}}}
}

private object SizeTracker {case class Sample(size: Long, numUpdates: Long)
}

...// Only use the last two samples to extrapolate// 如果sample太多了，就删除掉一些if (samples.size > 2) {samples.dequeue()}val bytesDelta = samples.toList.reverse match {case latest :: previous :: tail =>(latest.size - previous.size).toDouble / (latest.numUpdates - previous.numUpdates)// If fewer than 2 samples, assume no changecase _ => 0}bytesPerUpdate = math.max(0, bytesDelta)nextSampleNum = math.ceil(numUpdates * SAMPLE_GROWTH_RATE).toLong
}

// SizeTracker.scala
def estimateSize(): Long = {assert(samples.nonEmpty)val extrapolatedDelta = bytesPerUpdate * (numUpdates - samples.last.numUpdates)(samples.last.size + extrapolatedDelta).toLong
}

private[spark] abstract class ShuffleWriter[K, V] {/** Write a sequence of records to this task's output */@throws[IOException]def write(records: Iterator[Product2[K, V]]): Unit/** Close this writer, passing along whether the map completed */def stop(success: Boolean): Option[MapStatus]
}

// SortShuffleWriter
override def write(records: Iterator[Product2[K, V]]): Unit = {sorter = if (dep.mapSideCombine) {new ExternalSorter[K, V, C](context, dep.aggregator, Some(dep.partitioner), dep.keyOrdering, dep.serializer)} else {// 如果不需要进行mapSideCombine，那么我们传入空的aggregator和ordering，// 我们在map端不负责对key进行排序，统统留给reduce端吧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 mapOutputWriter = shuffleExecutorComponents.createMapOutputWriter(dep.shuffleId, mapId, dep.partitioner.numPartitions)...

//...sorter.writePartitionedMapOutput(dep.shuffleId, mapId, mapOutputWriter)val partitionLengths = mapOutputWriter.commitAllPartitions()

...mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths, mapId)
}// LocalDiskShuffleMapOutputWriter.java
@Override
public long[] commitAllPartitions() throws IOException {...cleanUp();File resolvedTmp = outputTempFile != null && outputTempFile.isFile() ? outputTempFile : null;blockResolver.writeIndexFileAndCommit(shuffleId, mapId, partitionLengths, resolvedTmp);return partitionLengths;
}

// IndexShuffleBlockResolver.java
def writeIndexFileAndCommit(shuffleId: Int, mapId: Long, lengths: Array[Long], dataTmp: File): Unit