3． 將DAG劃分為Stage剖析

DAG劃分Stage

一個Spark程序可以有多個DAG（有幾個Action，就有幾個DAG，上圖最后只有一個Action（圖中未表現(xiàn)），那么就是一個DAG）。

一個DAG可以有多個Stage（根據(jù)寬依賴／shuffle進行劃分）。

同一個Stage可以有多個Task并行執(zhí)行（task數(shù)＝分區(qū)數(shù)，如上圖，Stage1 中有三個分區(qū)P1、P2、P3，對應的也有三個 Task）。

可以看到這個DAG中只reduceByKey操作是一個寬依賴，Spark內(nèi)核會以此為邊界將其前后劃分成不同的Stage。

同時我們可以注意到，在圖中Stage1中，從textFile到flatMap到map都是窄依賴，這幾步操作可以形成一個流水線操作，通過flatMap操作生成的partition可以不用等待整個RDD計算結束，而是繼續(xù)進行map操作，這樣大大提高了計算的效率。

4． 提交Stages

調(diào)度階段的提交，最終會被轉換成一個任務集的提交，DAGScheduler通過TaskScheduler接口提交任務集，這個任務集最終會觸發(fā)TaskScheduler構建一個TaskSetManager的實例來管理這個任務集的生命周期，對于DAGScheduler來說，提交調(diào)度階段的工作到此就完成了。

而TaskScheduler的具體實現(xiàn)則會在得到計算資源的時候，進一步通過TaskSetManager調(diào)度具體的任務到對應的Executor節(jié)點上進行運算。

5． 監(jiān)控Job、Task、Executor

DAGScheduler監(jiān)控Job與Task：

要保證相互依賴的作業(yè)調(diào)度階段能夠得到順利的調(diào)度執(zhí)行，DAGScheduler需要監(jiān)控當前作業(yè)調(diào)度階段乃至任務的完成情況。

這通過對外暴露一系列的回調(diào)函數(shù)來實現(xiàn)的，對于TaskScheduler來說，這些回調(diào)函數(shù)主要包括任務的開始結束失敗、任務集的失敗，DAGScheduler根據(jù)這些任務的生命周期信息進一步維護作業(yè)和調(diào)度階段的狀態(tài)信息。

DAGScheduler監(jiān)控Executor的生命狀態(tài)：

TaskScheduler通過回調(diào)函數(shù)通知DAGScheduler具體的Executor的生命狀態(tài)，如果某一個Executor崩潰了，則對應的調(diào)度階段任務集的ShuffleMapTask的輸出結果也將標志為不可用，這將導致對應任務集狀態(tài)的變更，進而重新執(zhí)行相關計算任務，以獲取丟失的相關數(shù)據(jù)。

6． 獲取任務執(zhí)行結果

結果DAGScheduler：

一個具體的任務在Executor中執(zhí)行完畢后，其結果需要以某種形式返回給DAGScheduler，根據(jù)任務類型的不同，任務結果的返回方式也不同。

兩種結果，中間結果與最終結果：

對于FinalStage所對應的任務，返回給DAGScheduler的是運算結果本身。

而對于中間調(diào)度階段對應的任務ShuffleMapTask，返回給DAGScheduler的是一個MapStatus里的相關存儲信息，而非結果本身，這些存儲位置信息將作為下一個調(diào)度階段的任務獲取輸入數(shù)據(jù)的依據(jù)。

兩種類型，DirectTaskResult與IndirectTaskResult：

根據(jù)任務結果大小的不同，ResultTask返回的結果又分為兩類：

如果結果足夠小，則直接放在DirectTaskResult對象內(nèi)中。

如果超過特定尺寸則在Executor端會將DirectTaskResult先序列化，再把序列化的結果作為一個數(shù)據(jù)塊存放在BlockManager中，然后將BlockManager返回的BlockID放在IndirectTaskResult對象中返回給TaskScheduler，TaskScheduler進而調(diào)用TaskResultGetter將IndirectTaskResult中的BlockID取出并通過BlockManager最終取得對應的DirectTaskResult。

7． 任務調(diào)度總體詮釋

一張圖說明任務總體調(diào)度：

任務總體調(diào)度

Spark運行架構特點

 1． Executor進程專屬

每個Application獲取專屬的Executor進程，該進程在Application期間一直駐留，并以多線程方式運行Tasks。

Spark Application不能跨應用程序共享數(shù)據(jù)，除非將數(shù)據(jù)寫入到外部存儲系統(tǒng)。如圖所示：

Executor進程專屬

2． 支持多種資源管理器

Spark與資源管理器無關，只要能夠獲取Executor進程，并能保持相互通信就可以了。

Spark支持資源管理器包含：Standalone、On Mesos、On YARN、Or On EC2。如圖所示：

支持多種資源管理器

3． Job提交就近原則

提交SparkContext的Client應該靠近Worker節(jié)點（運行Executor的節(jié)點），最好是在同一個Rack（機架）里，因為Spark Application運行過程中SparkContext和Executor之間有大量的信息交換；

如果想在遠程集群中運行，最好使用RPC將SparkContext提交給集群，不要遠離Worker運行SparkContext。

如圖所示：

Job提交就近原則

4． 移動程序而非移動數(shù)據(jù)的原則執(zhí)行

移動程序而非移動數(shù)據(jù)的原則執(zhí)行，Task采用了數(shù)據(jù)本地性和推測執(zhí)行的優(yōu)化機制。

關鍵方法：taskIdToLocations、getPreferedLocations。

如圖所示：

數(shù)據(jù)本地性