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中方法來提交作業，在該方法中將進行劃分階段。