大数据：Spark Shuffle（二）Executor、Driver之间Shuffle结果消息传递、追踪

  override def statusUpdate(taskId: Long, state: TaskState, data: ByteBuffer) {val msg = StatusUpdate(executorId, taskId, state, data)driver match {case Some(driverRef) => driverRef.send(msg)case None => logWarning(s"Drop $msg because has not yet connected to driver")}}

• ExecutorId: Executor自己的ID
• TaskId: task分配的ID
• State: Task的运行状态

LAUNCHING, RUNNING, FINISHED, FAILED, KILLED, LOST

spark.driver.maxResultSize

spark.task.maxDirectResultSize

val serializedResult: ByteBuffer = {if (maxResultSize > 0 && resultSize > maxResultSize) {logWarning(s"Finished $taskName (TID $taskId). Result is larger than maxResultSize " +s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +s"dropping it.")ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))} else if (resultSize > maxDirectResultSize) {val blockId = TaskResultBlockId(taskId)env.blockManager.putBytes(blockId,new ChunkedByteBuffer(serializedDirectResult.duplicate()),StorageLevel.MEMORY_AND_DISK_SER)logInfo(s"Finished $taskName (TID $taskId). $resultSize bytes result sent via BlockManager)")ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))} else {logInfo(s"Finished $taskName (TID $taskId). $resultSize bytes result sent to driver")serializedDirectResult}}

case StatusUpdate(executorId, taskId, state, data) =>scheduler.statusUpdate(taskId, state, data.value)if (TaskState.isFinished(state)) {executorDataMap.get(executorId) match {case Some(executorInfo) =>executorInfo.freeCores += scheduler.CPUS_PER_TASKmakeOffers(executorId)case None =>// Ignoring the update since we don't know about the executor.logWarning(s"Ignored task status update ($taskId state $state) " +s"from unknown executor with ID $executorId")}}

 if (TaskState.isFinished(state)) {cleanupTaskState(tid)taskSet.removeRunningTask(tid)if (state == TaskState.FINISHED) {taskResultGetter.enqueueSuccessfulTask(taskSet, tid, serializedData)} else if (Set(TaskState.FAILED, TaskState.KILLED, TaskState.LOST).contains(state)) {taskResultGetter.enqueueFailedTask(taskSet, tid, state, serializedData)}}

 def enqueueSuccessfulTask(taskSetManager: TaskSetManager,tid: Long,serializedData: ByteBuffer): Unit = {getTaskResultExecutor.execute(new Runnable {override def run(): Unit = Utils.logUncaughtExceptions {try {val (result, size) = serializer.get().deserialize[TaskResult[_]](serializedData) match {case directResult: DirectTaskResult[_] =>if (!taskSetManager.canFetchMoreResults(serializedData.limit())) {return}// deserialize "value" without holding any lock so that it won't block other threads.// We should call it here, so that when it's called again in// "TaskSetManager.handleSuccessfulTask", it does not need to deserialize the value.directResult.value(taskResultSerializer.get())(directResult, serializedData.limit())case IndirectTaskResult(blockId, size) =>if (!taskSetManager.canFetchMoreResults(size)) {// dropped by executor if size is larger than maxResultSizesparkEnv.blockManager.master.removeBlock(blockId)return}logDebug("Fetching indirect task result for TID %s".format(tid))scheduler.handleTaskGettingResult(taskSetManager, tid)val serializedTaskResult = sparkEnv.blockManager.getRemoteBytes(blockId)if (!serializedTaskResult.isDefined) {/* We won't be able to get the task result if the machine that ran the task failed* between when the task ended and when we tried to fetch the result, or if the* block manager had to flush the result. */scheduler.handleFailedTask(taskSetManager, tid, TaskState.FINISHED, TaskResultLost)return}val deserializedResult = serializer.get().deserialize[DirectTaskResult[_]](serializedTaskResult.get.toByteBuffer)// force deserialization of referenced valuedeserializedResult.value(taskResultSerializer.get())sparkEnv.blockManager.master.removeBlock(blockId)(deserializedResult, size)}// Set the task result size in the accumulator updates received from the executors.// We need to do this here on the driver because if we did this on the executors then// we would have to serialize the result again after updating the size.result.accumUpdates = result.accumUpdates.map { a =>if (a.name == Some(InternalAccumulator.RESULT_SIZE)) {val acc = a.asInstanceOf[LongAccumulator]assert(acc.sum == 0L, "task result size should not have been set on the executors")acc.setValue(size.toLong)acc} else {a}}scheduler.handleSuccessfulTask(taskSetManager, tid, result)} catch {case cnf: ClassNotFoundException =>val loader = Thread.currentThread.getContextClassLoadertaskSetManager.abort("ClassNotFound with classloader: " + loader)// Matching NonFatal so we don't catch the ControlThrowable from the "return" above.case NonFatal(ex) =>logError("Exception while getting task result", ex)taskSetManager.abort("Exception while getting task result: %s".format(ex))}}})}

sched.dagScheduler.taskEnded(tasks(index), Success, result.value(), result.accumUpdates, info)

  def taskEnded(task: Task[_],reason: TaskEndReason,result: Any,accumUpdates: Seq[AccumulatorV2[_, _]],taskInfo: TaskInfo): Unit = {eventProcessLoop.post(CompletionEvent(task, reason, result, accumUpdates, taskInfo))}

  def compressSize(size: Long): Byte = {if (size == 0) {0} else if (size <= 1L) {1} else {math.min(255, math.ceil(math.log(size) / math.log(LOG_BASE)).toInt).toByte}}

  private[scheduler] def handleTaskCompletion(event: CompletionEvent) {
............
case smt: ShuffleMapTask =>val shuffleStage = stage.asInstanceOf[ShuffleMapStage]updateAccumulators(event)val status = event.result.asInstanceOf[MapStatus]val execId = status.location.executorIdlogDebug("ShuffleMapTask finished on " + execId)if (failedEpoch.contains(execId) && smt.epoch <= failedEpoch(execId)) {logInfo(s"Ignoring possibly bogus $smt completion from executor $execId")} else {shuffleStage.addOutputLoc(smt.partitionId, status)}if (runningStages.contains(shuffleStage) && shuffleStage.pendingPartitions.isEmpty) {markStageAsFinished(shuffleStage)logInfo("looking for newly runnable stages")logInfo("running: " + runningStages)logInfo("waiting: " + waitingStages)logInfo("failed: " + failedStages)// We supply true to increment the epoch number here in case this is a// recomputation of the map outputs. In that case, some nodes may have cached// locations with holes (from when we detected the error) and will need the// epoch incremented to refetch them.// TODO: Only increment the epoch number if this is not the first time//       we registered these map outputs.mapOutputTracker.registerMapOutputs(shuffleStage.shuffleDep.shuffleId,shuffleStage.outputLocInMapOutputTrackerFormat(),changeEpoch = true)clearCacheLocs()if (!shuffleStage.isAvailable) {// Some tasks had failed; let's resubmit this shuffleStage// TODO: Lower-level scheduler should also deal with thislogInfo("Resubmitting " + shuffleStage + " (" + shuffleStage.name +") because some of its tasks had failed: " +shuffleStage.findMissingPartitions().mkString(", "))submitStage(shuffleStage)} else {// Mark any map-stage jobs waiting on this stage as finishedif (shuffleStage.mapStageJobs.nonEmpty) {val stats = mapOutputTracker.getStatistics(shuffleStage.shuffleDep)for (job <- shuffleStage.mapStageJobs) {markMapStageJobAsFinished(job, stats)}}submitWaitingChildStages(shuffleStage)}}

  protected val mapStatuses = new ConcurrentHashMap[Int, Array[MapStatus]]().asScala
def registerMapOutputs(shuffleId: Int, statuses: Array[MapStatus], changeEpoch: Boolean = false) {mapStatuses.put(shuffleId, statuses.clone())if (changeEpoch) {incrementEpoch()}}

val blockFetcherItr = new ShuffleBlockFetcherIterator(context,blockManager.shuffleClient,blockManager,mapOutputTracker.getMapSizesByExecutorId(handle.shuffleId, startPartition, endPartition),// Note: we use getSizeAsMb when no suffix is provided for backwards compatibilitySparkEnv.get.conf.getSizeAsMb("spark.reducer.maxSizeInFlight", "48m") * 1024 * 1024,SparkEnv.get.conf.getInt("spark.reducer.maxReqsInFlight", Int.MaxValue))

def getMapSizesByExecutorId(shuffleId: Int, startPartition: Int, endPartition: Int): Seq[(BlockManagerId, Seq[(BlockId, Long)])] = {logDebug(s"Fetching outputs for shuffle $shuffleId, partitions $startPartition-$endPartition")val statuses = getStatuses(shuffleId)// Synchronize on the returned array because, on the driver, it gets mutated in placestatuses.synchronized {return MapOutputTracker.convertMapStatuses(shuffleId, startPartition, endPartition, statuses)}}

 private def getStatuses(shuffleId: Int): Array[MapStatus] = {val statuses = mapStatuses.get(shuffleId).orNullif (statuses == null) {logInfo("Don't have map outputs for shuffle " + shuffleId + ", fetching them")val startTime = System.currentTimeMillisvar fetchedStatuses: Array[MapStatus] = nullfetching.synchronized {// Someone else is fetching it; wait for them to be donewhile (fetching.contains(shuffleId)) {try {fetching.wait()} catch {case e: InterruptedException =>}}// Either while we waited the fetch happened successfully, or// someone fetched it in between the get and the fetching.synchronized.fetchedStatuses = mapStatuses.get(shuffleId).orNullif (fetchedStatuses == null) {// We have to do the fetch, get others to wait for us.fetching += shuffleId}}if (fetchedStatuses == null) {// We won the race to fetch the statuses; do sologInfo("Doing the fetch; tracker endpoint = " + trackerEndpoint)// This try-finally prevents hangs due to timeouts:try {val fetchedBytes = askTracker[Array[Byte]](GetMapOutputStatuses(shuffleId))fetchedStatuses = MapOutputTracker.deserializeMapStatuses(fetchedBytes)logInfo("Got the output locations")mapStatuses.put(shuffleId, fetchedStatuses)} finally {fetching.synchronized {fetching -= shuffleIdfetching.notifyAll()}}}logDebug(s"Fetching map output statuses for shuffle $shuffleId took " +s"${System.currentTimeMillis - startTime} ms")if (fetchedStatuses != null) {return fetchedStatuses} else {logError("Missing all output locations for shuffle " + shuffleId)throw new MetadataFetchFailedException(shuffleId, -1, "Missing all output locations for shuffle " + shuffleId)}} else {return statuses}}

• 封装了一层缓存mapStatus，对同一个Executor来说，里面的线程都是运行同一个Driver的提交的任务，对相同的shuffeID，MapStatus是一样的
• 对同一个Executor、ShuffeID来说，通过Driver获取信息只需要一次，Driver里保存的Shuffle的结果是单点的，对同一个Executor来说获取同一个ShuffleID只需要请求一次，在Traker里面保存了一个队列fetching，里面保存的ShuffeID代表的是有线程正在从Driver端获取ShuffleID的MapStatus，如果发现有值，当前线程会等待，直到其他的线程获取ShuffleID状态并保存到缓存结束，当前线程直接从缓存中获取当前状态
• Executor 向Driver发送GetMapOutputStatuses(shuffleId)消息
• Driver收到GetMapOutputStatuses消息后保存到消息队列mapOutputRequests，Map-Output-Dispatcher-x多线程处理消息队列，返回序列化的MapStatus
• Executor反序列化成MapStatus

 private def convertMapStatuses(shuffleId: Int,startPartition: Int,endPartition: Int,statuses: Array[MapStatus]): Seq[(BlockManagerId, Seq[(BlockId, Long)])] = {assert (statuses != null)val splitsByAddress = new HashMap[BlockManagerId, ArrayBuffer[(BlockId, Long)]]for ((status, mapId) <- statuses.zipWithIndex) {if (status == null) {val errorMessage = s"Missing an output location for shuffle $shuffleId"logError(errorMessage)throw new MetadataFetchFailedException(shuffleId, startPartition, errorMessage)} else {for (part <- startPartition until endPartition) {splitsByAddress.getOrElseUpdate(status.location, ArrayBuffer()) +=((ShuffleBlockId(shuffleId, mapId, part), status.getSizeForBlock(part)))}}}splitsByAddress.toSeq}