spark 作业执行原理源码阅读(三)
概述
作业(Job)
调度阶段(stage)
任务(Task)
DAGScheduler:面向调度阶段的任务调节器,负责接收spark应用提交的作业,根据RDD的依赖关系(根据宽依赖划分)划分调度阶段,并提交stage给TaskScheduler。
TaskScheduler:面向任务的调度器,接收DAGScheduler提交过来的stage,然后以stage划分后的结果,将Task分发到Work节点运行,有Worker节点的Exectuor来运行该任务。
Spark中对RDD的操作算子主要分为转换操作和行动操作,对于转换操作是lazy级别的,也就是延迟执行,只有出现了行动操作才出发了作业的提交。
1、Job由DAGScheduler划分成多个stage
2、stage由TaskScheduler划分成多个Task
3、每个TaskScheduler只为一个SparkContext实例服务,TaskScheduler接收来自DAGScheduler过来的任务集,把任务集以任务的形式一个一个分发到集群Worker节点的Executor中去运行。如果某个任务运行失败,TaskScheduler要负责重试,如果发现其中一个任务一直为运行完,就可能启动同样的任务,哪个先运行完就用哪个结果。
4、Worker中的Executor收到TaskScheduler发送的任务后,以多线程的方式运行,每一个线程负责一个任务,运行结束后要返回给TaskScheduler,不同类型的任务,返回的方式也不同,shuffleMapTask返回的是一个MapsStatus对象,而不是结果本身,ResultTask根据大小的冉,返回方式也可分为两类。
5、下图展示了作业和任务调度方法之间的调用关系,在类之间既有直接调用,也有通过RPC远程调用,在途中使用了虚线箭头进行标记的即为远程调用。
分段
提交作业
因为转换操作是延迟执行,所以作业真正提交是从"count"这个行动操作出现开始的,在RDD的源码中,count方法触发了SparkContext的runJob方法来提交作业,这部分是由内部隐形调用runJob,不需要用户显性的去提交作业。
对于RDD,会根据依赖关系形成一个有向无环图(DAG),然后把DAG交给DAGScheduler来处理。SparkContext的runJob方法经过几次调用后,进入DAGScheduler的runJob方法,其中DAGScheduler的runJob方法如下:
/**
* Run an action job on the given RDD and pass all the results to the resultHandler function as
* they arrive.
*
* @param rdd target RDD to run tasks on
* @param func a function to run on each partition of the RDD
* @param partitions set of partitions to run on; some jobs may not want to compute on all
* partitions of the target RDD, e.g. for operations like first()
* @param callSite where in the user program this job was called
* @param resultHandler callback to pass each result to
* @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
*
* @throws Exception when the job fails
*/
def runJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): Unit = {
val start = System.nanoTime
val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
// Note: Do not call Await.ready(future) because that calls `scala.concurrent.blocking`,
// which causes concurrent SQL executions to fail if a fork-join pool is used. Note that
// due to idiosyncrasies in Scala, `awaitPermission` is not actually used anywhere so it's
// safe to pass in null here. For more detail, see SPARK-13747.
val awaitPermission = null.asInstanceOf[scala.concurrent.CanAwait]
waiter.completionFuture.ready(Duration.Inf)(awaitPermission)
waiter.completionFuture.value.get match {
case scala.util.Success(_) =>
logInfo("Job %d finished: %s, took %f s".format
(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
case scala.util.Failure(exception) =>
logInfo("Job %d failed: %s, took %f s".format
(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
// SPARK-8644: Include user stack trace in exceptions coming from DAGScheduler.
val callerStackTrace = Thread.currentThread().getStackTrace.tail
exception.setStackTrace(exception.getStackTrace ++ callerStackTrace)
throw exception
}
}
DAGScheduler会进行一系列的方法调用:
1、在runJob中调用submitJob方法来继续提交作业,这里会发生阻塞,直到作业返回结果,成功或者失败。
2、在submitJob中创建了JobWaiter对象,借助内部消息处理机制,把该对象传送给内置对象DAGSchedulerEventProcessLoop进行处理
3、在DAGSchedulerEventProcessLoop消息接收方法OnReceive中,接收到Jobsumitted样例完成模式匹配后,继续调用DAGScheduler的handleJobSubmitted中方法来提交作业,在该方法中将进行划分阶段。