[Spark源码剖析] DAGScheduler提交stage

DAGScheduler通过调用submitStage来提交stage,实现如下:

  private def submitStage(stage: Stage) {
    val jobId = activeJobForStage(stage)
    if (jobId.isDefined) {
      logDebug("submitStage(" + stage + ")")
      if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
        //< 获取该stage未提交的父stages,并按stage id从小到大排序
        val missing = getMissingParentStages(stage).sortBy(_.id)
        logDebug("missing: " + missing)
        if (missing.isEmpty) {
          logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
          //< 若无未提交的父stage, 则提交该stage对应的tasks
          submitMissingTasks(stage, jobId.get)
        } else {
          //< 若存在未提交的父stage, 依次提交所有父stage (若父stage也存在未提交的父stage, 则提交之, 依次类推); 并把该stage添加到等待stage队列中
          for (parent <- missing) {
            submitStage(parent)
          }
          waitingStages += stage
        }
      }
    } else {
      abortStage(stage, "No active job for stage " + stage.id)
    }
  }

submitStage先调用getMissingParentStages来获取参数stageX(这里为了区分,取名为stageX)是否有未提交的父stages,若有,则依次递归(按stage id从小到大排列,也就是stage是从后往前提交的)提交父stages,并将stageX加入到waitingStages: HashSet[Stage]中。对于要依次提交的父stage,也是如此。

getMissingParentStagesDAGScheduler划分stage中介绍的getParentStages有点像,但不同的是不再需要划分stage,并对每个stage的状态做了判断,源码及注释如下:

//< 以参数stage为起点,向前遍历所有stage,判断stage是否为未提交,若使则加入missing中
  private def getMissingParentStages(stage: Stage): List[Stage] = {
    //< 未提交的stage
    val missing = new HashSet[Stage]
    //< 存储已经被访问到得RDD
    val visited = new HashSet[RDD[_]]

    val waitingForVisit = new Stack[RDD[_]]
    def visit(rdd: RDD[_]) {
      if (!visited(rdd)) {
        visited += rdd
        if (getCacheLocs(rdd).contains(Nil)) {
          for (dep <- rdd.dependencies) {
            dep match {
              //< 若为宽依赖,生成新的stage
              case shufDep: ShuffleDependency[_, _, _] =>
                //< 这里调用getShuffleMapStage不像在getParentStages时需要划分stage,而是直接根据shufDep.shuffleId获取对应的ShuffleMapStage
                val mapStage = getShuffleMapStage(shufDep, stage.jobId)
                if (!mapStage.isAvailable) {
                  // 若stage得状态为available则为未提交stage
                  missing += mapStage
                }
              //< 若为窄依赖,那就属于同一个stage。并将依赖的RDD放入waitingForVisit中,以能够在下面的while中继续向上visit,直至遍历了整个DAG图
              case narrowDep: NarrowDependency[_] =>
                waitingForVisit.push(narrowDep.rdd)
            }
          }
        }
      }
    }
    waitingForVisit.push(stage.rdd)
    while (waitingForVisit.nonEmpty) {
      visit(waitingForVisit.pop())
    }
    missing.toList
  }

上面提到,若stageX存在未提交的父stages,则先提交父stages;那么,如果stageX没有未提交的父stage呢(比如,包含从HDFS读取数据生成HadoopRDD的那个stage是没有父stage的)?

这时会调用submitMissingTasks(stage, jobId.get),参数就是stageX及其对应的jobId.get。这个函数便是我们时常在其他文章或书籍中看到的将stage与taskSet对应起来,然后DAGScheduler将taskSet提交给TaskScheduler去执行的实施者。这个函数的实现比较长,下面分段说明。

Step1: 得到RDD中需要计算的partition

对于Shuffle类型的stage,需要判断stage中是否缓存了该结果;对于Result类型的Final Stage,则判断计算Job中该partition是否已经计算完成。这么做(没有直接提交全部tasks)的原因是,stage中某个task执行失败其他执行成功的时候就需要找出这个失败的task对应要计算的partition而不是要计算所有partition

  private def submitMissingTasks(stage: Stage, jobId: Int) {
    stage.pendingTasks.clear()

    //< 首先得到RDD中需要计算的partition
    //< 对于Shuffle类型的stage,需要判断stage中是否缓存了该结果;
    //< 对于Result类型的Final Stage,则判断计算Job中该partition是否已经计算完成
    //< 这么做的原因是,stage中某个task执行失败其他执行成功地时候就需要找出这个失败的task对应要计算的partition而不是要计算所有partition
    val partitionsToCompute: Seq[Int] = {
      stage match {
        case stage: ShuffleMapStage =>
          (0 until stage.numPartitions).filter(id => stage.outputLocs(id).isEmpty)
        case stage: ResultStage =>
          val job = stage.resultOfJob.get
          (0 until job.numPartitions).filter(id => !job.finished(id))
      }
    }

Step2: 序列化task的binary

Executor可以通过广播变量得到它。每个task运行的时候首先会反序列化

var taskBinary: Broadcast[Array[Byte]] = null
    try {
      // For ShuffleMapTask, serialize and broadcast (rdd, shuffleDep).
      // For ResultTask, serialize and broadcast (rdd, func).
      val taskBinaryBytes: Array[Byte] = stage match {
        case stage: ShuffleMapStage =>
          //< 对于ShuffleMapTask,将rdd及其依赖关系序列化;在Executor执行task之前会反序列化
          closureSerializer.serialize((stage.rdd, stage.shuffleDep): AnyRef).array()
          //< 对于ResultTask,对rdd及要在每个partition上执行的func
        case stage: ResultStage =>
          closureSerializer.serialize((stage.rdd, stage.resultOfJob.get.func): AnyRef).array()
      }

      //< 将序列化好的信息广播给所有的executor
      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)
        runningStages -= stage

        // Abort execution
        return
      case NonFatal(e) =>
        abortStage(stage, s"Task serialization failed: $e\n${e.getStackTraceString}")
        runningStages -= stage
        return
    }

Step3: 为每个需要计算的partiton生成一个task

ShuffleMapStage对应的task全是ShuffleMapTask; ResultStage对应的全是ResultTask。task继承Serializable,要确保task是可序列化的。

val tasks: Seq[Task[_]] = stage match {
      case stage: ShuffleMapStage =>
        partitionsToCompute.map { id =>
          val locs = getPreferredLocs(stage.rdd, id)
          //< RDD对应的partition
          val part = stage.rdd.partitions(id)
          new ShuffleMapTask(stage.id, taskBinary, part, locs)
        }

      case stage: ResultStage =>
        val job = stage.resultOfJob.get
        //< id为输出分区索引,表示reducerID
        partitionsToCompute.map { id =>
          val p: Int = job.partitions(id)
          val part = stage.rdd.partitions(p)
          val locs = getPreferredLocs(stage.rdd, p)
          new ResultTask(stage.id, taskBinary, part, locs, id)
        }
    }

Step4: 提交tasks

先用tasks来初始化一个TaskSet对象,再调用TaskScheduler.submitTasks提交

stage.pendingTasks ++= tasks
      logDebug("New pending tasks: " + stage.pendingTasks)
      //< 提交TaskSet至TaskScheduler
      taskScheduler.submitTasks(
        new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.jobId, properties))
      //< 记录stage提交task的时间
      stage.latestInfo.submissionTime = Some(clock.getTimeMillis())
    } else {

以上,介绍了提交stage和提交tasks的实现。本文若有纰漏,请批评指正。


本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏浪淘沙

KafKa 代码实现

23330
来自专栏about云

spark2的SparkSession思考与总结2:SparkSession有哪些函数及作用是什么

问题导读 1.spark SparkSession包含哪些函数? 2.创建DataFrame有哪些函数? 3.创建DataSet有哪些函数? 上一篇spa...

81550
来自专栏coolblog.xyz技术专栏

Spring AOP 源码分析 - 创建代理对象

在上一篇文章中,我分析了 Spring 是如何为目标 bean 筛选合适的通知器的。现在通知器选好了,接下来就要通过代理的方式将通知器(Advisor)所持有的...

26270
来自专栏码匠的流水账

聊聊sentinel的SimpleHttpCommandCenter

sentinel-transport-simple-http-0.1.1-sources.jar!/com/alibaba/csp/sentinel/trans...

10010
来自专栏菩提树下的杨过

Jboss EAP:native management API学习

上一节已经学习了CLI命令行来控制JBOSS,如果想在程序中以编码方式来控制JBOSS,可以参考下面的代码,实际上在前面的文章,用代码控制Jboss上的Data...

22290
来自专栏java达人

Spring aop 的代理机制

Spring aop 是通过代理实现的,代理有静态代理,jdk动态代理和cglib动态代理,代理就像我们生活中的房产中介,你不直接与房主,银行接触,而是通过中介...

20920
来自专栏JavaEdge

Mybatis#BaseExecutor源码解析BaseExecutor源码解析

BaseExecutor是Executor的一个子类,是一个抽象类,实现接口Executor的部分方法,并提供了三个抽象方法

13560
来自专栏Java帮帮-微信公众号-技术文章全总结

Java面试系列17-编程题-读取服务器字符、实现序列化、计数器、1000阶乘、n出列问题等

一,Java的通信编程,编程题(或问答),用JAVA SOCKET编程,读服务器几个字符,再写入本地显示? Server端程序: package test;...

48280
来自专栏java达人

Spring aop 的代理机制

Spring aop 是通过代理实现的,代理有静态代理,jdk动态代理和cglib动态代理,代理就像我们生活中的房产中介,你不直接与房主,银行接触,而是通过中介...

21390
来自专栏微信公众号:Java团长

Java类加载器详解(下)

这个类中定义了一个加密和解密的算法,很简单的,就是将字节和oxff异或一下即可,而且这个算法是加密和解密的都可以用,很是神奇呀!

16730

扫码关注云+社区

领取腾讯云代金券