Spark DAG调度

SparkContext在初始化时,创建了DAG调度与Task调度来负责RDD Action操作的调度执行。

DAGScheduler

DAGScheduler负责Spark的最高级别的任务调度,调度的粒度是Stage,它为每个Job的所有Stage计算一个有向无环图,控制它们的并发,并找到一个最佳路径来执行它们。具体的执行过程是将Stage下的Task集提交给TaskScheduler对象,由它来提交到集群上去申请资源并最终完成执行。

DAGScheduler的定义位于scheduler/DAGScheduler.scala中。下面是它的类声明,初始化时除了需要一个SparkContext对象外,最重要的是需要输入一个TaskScheduler对象来负责Task的执行:

private[spark]
class DAGScheduler(
    private[scheduler] val sc: SparkContext,
    private[scheduler] val taskScheduler: TaskScheduler,
    listenerBus: LiveListenerBus,
    mapOutputTracker: MapOutputTrackerMaster,
    blockManagerMaster: BlockManagerMaster,
    env: SparkEnv,
    clock: Clock = new SystemClock())
extends Logging {
    ...
}
1.runJob过程

所有需要执行的RDD Action,都会调用SparkContext.runJob来提交任务,而SparkContext.runJob调用的是DAGScheduler.runJob。下面是runJob的定义:

def runJob[T, U](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) => U,
    partitions: Seq[Int],
    callSite: CallSite,
    allowLocal: Boolean,
    resultHandler: (Int, U) => Unit,
    properties: Properties): Unit = {
    val start = System.nanoTime
    val waiter = submitJob(rdd, func, partitions, callSite, allowLocal,
        resultHandler, properties)
    waiter.awaitResult() match {
        case JobSucceeded =>
            logInfo("Job %d finished: %s, took %f s".format
                (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
        case JobFailed(exception: Exception) =>
            logInfo("Job %d failed: %s, took %f s".format
                (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
        throw exception
    }
}

runJob调用submitJob提交任务,并等待任务结束。

任务提交后的处理过程大概如下。

(1) submitJob生成新的Job ID,发送消息JobSubmitted

(2) DAG收到JobSubmitted消息,调用handleJobSubmitted来处理。

(3) handleJobSubmitted创建一个ResultStage,并使用submitStage来提交这个ResultStage

上面的过程看起来没完,实际上大的过程已经结束了,猫腻在submitStage中。Spark的执行过程是“懒惰”(lazy)的,这在这里得到了完整的体现。任务提交时,不是按Job的先后顺序提交的,而是倒序的。每个Job的最后一个操作是Action操作,DAG把这最后的Action操作当作一个Stage,首先提交,然后逆向逐级递归填补缺少的上级Stage,从而生成一棵实现最后Action操作的最短的(因为都是必须的)有向无环图,然后再从头开始计算。

submitStage方法的实现代码如下所示:

// 提交Stage之前,先递归提交所缺失的父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) {
                // 仅在所有缺失的父Stage都提交执行了,才开始提交自己
                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)
    }
}

可以看到,这是一个逆向递归的过程,先查找所有缺失的上级Stage并提交,待所有上级Stage都提交执行了,才轮到执行当前Stage对应的Task。

查找上级Stage的过程,其实就是递归向上遍历所有RDD依赖列表并生成Stage的过程,代码如下所示:

private def getMissingParentStages(stage: Stage): List[Stage] = {
    val missing = new HashSet[Stage]
    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 {
                        case shufDep: ShuffleDependency[_, _, _] =>
                            val mapStage = getShuffleMapStage(shufDep, stage.jobId)
                            if (!mapStage.isAvailable) {
                                missing += mapStage
                            }
                        case narrowDep: NarrowDependency[_] =>
                            waitingForVisit.push(narrowDep.rdd)
                    }
                }
            }
        }
    }
    waitingForVisit.push(stage.rdd)
    while (waitingForVisit.nonEmpty) {
        visit(waitingForVisit.pop())
    }
    missing.toList
}

熟悉数据结构的读者不难发现,遍历的过程是非递归的层序遍历(不是前序、中序或后序),使用了一个堆栈来协助遍历,而且保证了层序的顺序与DAG中的依赖顺序一致。

2.Stage

值得注意的是,仅对依赖类型是ShuffleDependency的RDD操作创建Stage,其他的RDD操作并没有创建Stage。这里我们补一下Stage的概念。前面讲RDD时提到过,RDD操作有两类依赖:一类是窄依赖,一个RDD分区只依赖上一个RDD的部分分区,而且这些分区都在相同的节点上;另外一类依赖是Shuffle依赖,一个RDD分区可能会依赖上一级RDD的全部分区,一个典型的例子是groupBy聚合操作。这两类操作在计算上有明显的区别,窄依赖都在同一个节点上进行计算,而Shuffle依赖跨越多个节点,甚至所有涉及的计算节点。因此,DAG在调度时,对于在相同节点上进行的Task计算,会合并为一个Stage。

所以,只有两种情况下会生成新的Stage:一类是依赖类型是Shuffle的Transformation操作会触发生成新的Stage,几乎所有的ByKey操作都是,比如reduceByKeygroupByKey;另外一类是Action操作,是为了生成默认的Stage,这样即便没有Shuffle类操作,保证至少有一个Stage。

总结一下,各Stage之间以Shuffle为分界线。

TaskScheduler

相对DAGScheduler而言,TaskScheduler是低级别的调度接口,允许实现不同的Task调度器。目前,已经实现的Task调度器除了自带的以外,还有YARN和Mesos调度器。每个TaskScheduler对象只服务于一个SparkContext的Task调度。TaskSchedulerDAGScheduler的每个Stage接收一组Task,并负责将它们发送到集群上,运行它们,如果出错还会重试,最后返回消息给DAGScheduler

TaskScheduler的主要接口包括一个钩子接口(也称hook,表示定义好之后,不是用户主动调用的),被调用的时机是在初始化完成之后和调度启动之前:

def postStartHook() { }

还有启动和停止调度的命令:

def start(): Unit
def stop(): Unit

此外,还有提交和撤销Task集的命令:

def submitTasks(taskSet: TaskSet): Unit
def cancelTasks(stageId: Int, interruptThread: Boolean)

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏大数据-Hadoop、Spark

Spark Streaming + Kafka整合

2185
来自专栏码匠的流水账

聊聊jdbc的batch操作

statement的batch操作,可以批量进行insert或update操作,提升操作性能,特别是在大数据量的insert或update的时候。

712
来自专栏Hadoop实操

Spark2Streaming读Kerberos环境的Kafka并写数据到HDFS

在前面的文章Fayson介绍了一些关于Spark2Streaming的示例如《Spark2Streaming读Kerberos环境的Kafka并写数据到HBas...

1371
来自专栏Hadoop实操

Spark2Streaming读Kerberos环境的Kafka并写数据到Hive

在前面的文章Fayson介绍了一些关于Spark2Streaming的示例如《Spark2Streaming读Kerberos环境的Kafka并写数据到HBas...

4333
来自专栏个人分享

Shuffle相关分析

 Shuffle描述是一个过程,表现出的是多对多的依赖关系。Shuffle是连接map阶段和Reduce阶段的纽带,每个Reduce Task都会从Map Ta...

863
来自专栏大数据和云计算技术

SparkStreaming入门

黄文辉同学第二篇,请大家支持! 1.SparkStreaming简介 Spark Streaming属于核心Spark API的扩展,支持实时数据流的可扩展、高...

3254
来自专栏个人分享

Spark作业调度

    Spark在任务提交时,主要存在于Driver和Executor的两个节点.

991
来自专栏Albert陈凯

Spark系列课程-00xxSpark RDD持久化

我们这节课讲一下RDD的持久化 ? RDD的持久化 这段代码我们上午已经看过了,有瑕疵大家看出来了吗? 有什么瑕疵啊? 大家是否还记得我在第二节课的时候跟大...

3778
来自专栏Java 源码分析

SparkStreaming 入门

2158
来自专栏Albert陈凯

3.4 RDD的计算

3.4 RDD的计算 3.4.1 Ta s k简介 原始的RDD经过一系列转换后,会在最后一个RDD上触发一个动作,这个动作会生成一个Job。在Job被划分为...

34410

扫码关注云+社区