3． 將DAG劃分為Stage剖析

DAG劃分Stage

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

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

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

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

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

4． 提交Stages

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

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

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

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

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

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

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

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

6． 獲取任務(wù)執(zhí)行結(jié)果

結(jié)果DAGScheduler：

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

兩種結(jié)果，中間結(jié)果與最終結(jié)果：

對于FinalStage所對應(yīng)的任務(wù)，返回給DAGScheduler的是運(yùn)算結(jié)果本身。

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

兩種類型，DirectTaskResult與IndirectTaskResult：

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

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

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

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

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

任務(wù)總體調(diào)度

Spark運(yùn)行架構(gòu)特點(diǎn)

 1． Executor進(jìn)程專屬

每個(gè)Application獲取專屬的Executor進(jìn)程，該進(jìn)程在Application期間一直駐留，并以多線程方式運(yùn)行Tasks。

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

Executor進(jìn)程專屬

2． 支持多種資源管理器

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

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

支持多種資源管理器

3． Job提交就近原則

提交SparkContext的Client應(yīng)該靠近Worker節(jié)點(diǎn)（運(yùn)行Executor的節(jié)點(diǎn)），最好是在同一個(gè)Rack（機(jī)架）里，因?yàn)镾park Application運(yùn)行過程中SparkContext和Executor之間有大量的信息交換；

如果想在遠(yuǎn)程集群中運(yùn)行，最好使用RPC將SparkContext提交給集群，不要遠(yuǎn)離Worker運(yùn)行SparkContext。

如圖所示：

Job提交就近原則

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

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

關(guān)鍵方法：taskIdToLocations、getPreferedLocations。

如圖所示：

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