首页 > 其他分享> > spark 上下游shuffle结果的存放获取

spark 上下游shuffle结果的存放获取

2020-01-25 10:04:17 作者：互联网

当一个job在DAGScheduler中被分隔为stage，将会根据其内部的shuffle关系将整个job整理出ShuffleMapStage，而最后结果的ResultStage在提交时，将会不断遍历其parent stage，而本身被加入DAGScheduler的等待集合，只在所有parent的stage执行完毕之后才会执行任务流程中的child stage。

private def submitStage(stage: Stage) {
  val jobId = activeJobForStage(stage)
  if (jobId.isDefined) {
    logDebug("submitStage(" + stage + ")")
    if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
      val missing = getMissingParentStages(stage).sortBy(_.id)
      logDebug("missing: " + missing)
      if (missing.isEmpty) {
        logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
        submitMissingTasks(stage, jobId.get)
      } else {
        for (parent <- missing) {
          submitStage(parent)
        }
        waitingStages += stage
      }
    }
  } else {
    abortStage(stage, "No active job for stage " + stage.id, None)
  }
}

在上方的代码可以看到，当子stage提交之后，将会不断递归，将parent stage调用该方法进行提交，在这里，最上级的stage将会被通过submitMissingTasks()方法被分成task进行执行，而下游的stage将会放入等待集合，直到上级全部执行完毕才回进入执行计划。

通过submitMissingTasks()方法，将会把当前需要执行的stage转换成task交给Executor执行，ShuffleStage也将会被转换成ShuffleMapTask经由Executor执行。

setTaskFinishedAndClearInterruptStatus()
execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)

在Executor执行完ShuffleMapTask的结尾，将会把ShuffleMapTask执行完毕的结果写到BlockManager中，并把其在BlockManager中的地址整理成一份map，作为task执行完毕的结果序列化后返回告知driver端。

在driver端，task的执行结果，尤其是ShuffleMapTask的结果的处理，主要分为两步，一步是注册shuffle结果在BlockManager中的位置，另一个便是启动其下游的child stage。

第一步，将会通过一个MapOutputTracker的成员实现。

mapOutputTracker.registerMapOutput(
  shuffleStage.shuffleDep.shuffleId, smt.partitionId, status)

在处理task处理结果的时候，ShuffleMapTask会通过MapOutputTracker的registerMapOutput()方法将其shuffleid以及分区号和处理结果存放到处于driver端的MapOutputTracker中。

MapOutputTracker维护着一个map，以ShuffleId作为key用来保存shuffle处理结果的一个映射，以便当需要具体使用到的时候可以快速获取其处理结果在BlockManager上的位置。

第二步，在其处理完之后，将会调用submitWaitingChildStages()方法尝试将child stage进行执行。

private def submitWaitingChildStages(parent: Stage) {
  logTrace(s"Checking if any dependencies of $parent are now runnable")
  logTrace("running: " + runningStages)
  logTrace("waiting: " + waitingStages)
  logTrace("failed: " + failedStages)
  val childStages = waitingStages.filter(_.parents.contains(parent)).toArray
  waitingStages --= childStages
  for (stage <- childStages.sortBy(_.firstJobId)) {
    submitStage(stage)
  }
}

将会从等待集合获取该stage的所有子stage准备通过本文最一开始的submitStage()方法调用，但是在submitStage()将会再次判断，直到其所有parent全部执行完才会正式执行。

Child stage如何获取上述的shuffle 结果？

以ShuffleRowRDD的compute()方法为例子。

override def compute(split: Partition, context: TaskContext): Iterator[InternalRow] = {
  val shuffledRowPartition = split.asInstanceOf[ShuffledRowRDDPartition]
  // The range of pre-shuffle partitions that we are fetching at here is
  // [startPreShufflePartitionIndex, endPreShufflePartitionIndex - 1].
  val reader =
    SparkEnv.get.shuffleManager.getReader(
      dependency.shuffleHandle,
      shuffledRowPartition.startPreShufflePartitionIndex,
      shuffledRowPartition.endPreShufflePartitionIndex,
      context)
  reader.read().asInstanceOf[Iterator[Product2[Int, InternalRow]]].map(_._2)
}

其将会直接构造一个reader尝试读取shuffle的结果。

最后构造的实际上是一个BlockStoreShuffleReader，而顾名思义，shuffle的结果也存放在BlockManager当中。

val wrappedStreams = new ShuffleBlockFetcherIterator(
  context,
  blockManager.shuffleClient,
  blockManager,
  mapOutputTracker.getMapSizesByExecutorId(handle.shuffleId, startPartition, endPartition),
  serializerManager.wrapStream,
  // Note: we use getSizeAsMb when no suffix is provided for backwards compatibility
  SparkEnv.get.conf.getSizeAsMb("spark.reducer.maxSizeInFlight", "48m") * 1024 * 1024,
  SparkEnv.get.conf.getInt("spark.reducer.maxReqsInFlight", Int.MaxValue),
  SparkEnv.get.conf.get(config.REDUCER_MAX_BLOCKS_IN_FLIGHT_PER_ADDRESS),
  SparkEnv.get.conf.get(config.MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM),
  SparkEnv.get.conf.getBoolean("spark.shuffle.detectCorrupt", true))

在其read()方法中，将会构造一个数据流，而数据流的获取目的地坐标，则是通过MapOutputTracker的getMapSizesByExecutorId()方法来获取。

当executor这个方法被调用的时候，首先会通过getStatus()尝试和master获取具体的存放数据，如果不存在将会构造网络请求向driver端MapOutputTracker尝试拉取对应的shuffleId的具体存放位置。当获取之后，shuffle的具体数据也将得到，可以构造数据流的形式作为child stage的数据继续执行。

tydhot 发布了141 篇原创文章 · 获赞 19 · 访问量 10万+ 私信关注

标签：上下游,parent,get,将会,shuffle,spark,执行,stage
来源： https://blog.csdn.net/weixin_40318210/article/details/104082535