spark調(diào)度管理原理

1、Spark調(diào)度管理原理 Spark調(diào)度管理原理 Spark程序得以運(yùn)轉(zhuǎn)的核心是它的調(diào)度管理邏輯，在Spark底層物理資源調(diào)度到上層應(yīng)用。處于核心位置的作業(yè)調(diào)度管理模塊是整個(gè)程序運(yùn)作的樞紐。4.1 Spark作業(yè)調(diào)度管理概述 在Spark作業(yè)調(diào)度系統(tǒng)中，調(diào)度的前提是判斷多個(gè)作業(yè)任務(wù)的依賴關(guān)系，這些作業(yè)任務(wù)之間可能存在因果的依賴關(guān)系。適合用DAG有向無環(huán)圖來表示。 在作業(yè)調(diào)度的相關(guān)類中，最重要的就是DAGScheduler（基于DAG圖的調(diào)度類）。 TaskScheduler負(fù)責(zé)每個(gè)具體任務(wù)的實(shí)際物理調(diào)度，DAGScheduler負(fù)責(zé)將作業(yè)拆分成不同階段的具有依賴關(guān)系的多批任務(wù)，可以理解為DAGS

2、cheduler負(fù)責(zé)任務(wù)的邏輯調(diào)度。4.2 Spark調(diào)度的相關(guān)基本概念Task（任務(wù)）：?jiǎn)蝹€(gè)分區(qū)數(shù)據(jù)集上的最小處理流程單元。TaskSet （任務(wù)集）：由一組關(guān)聯(lián)的，但相互之間沒有Shuffle依賴關(guān)系的任務(wù)所組成的任務(wù)集。Stage（調(diào)度階段）：一個(gè)任務(wù)集對(duì)應(yīng)的調(diào)度階段。Job（作業(yè)）：由一個(gè)或多個(gè)調(diào)度階段所組成的一次計(jì)算作業(yè)。Application（應(yīng)用程序）：Spark應(yīng)用程序，由一個(gè)或多個(gè)作業(yè)組成。 4.3 作業(yè)調(diào)度模塊頂層邏輯概述 在Spark應(yīng)用程序中，用戶代碼基本是基于RDD的一系列計(jì)算操作。在實(shí)際運(yùn)行時(shí)，這些計(jì)算操作是延遲執(zhí)行的，并不是所有的RDD操作都會(huì)觸發(fā)Spark向集群

3、提交實(shí)際作業(yè)，只有一些需要返回?cái)?shù)據(jù)或者向外部輸出數(shù)據(jù)的操作才會(huì)觸發(fā)實(shí)際的計(jì)算工作，其他的變換操作只是生成對(duì)應(yīng)的RDD關(guān)系鏈，用來記錄依賴關(guān)系和所需執(zhí)行的運(yùn)算。 DAGScheduler最重要的任務(wù)之一就是計(jì)算作業(yè)和任務(wù)的依賴關(guān)系，制定調(diào)度邏輯。DAGScheduler中作業(yè)調(diào)度的發(fā)起，對(duì)外暴露的兩個(gè)主要的入口其中之一是submitJob，另一個(gè)是runJob。兩者的區(qū)別在于前者返回一個(gè)Jobwaiter對(duì)象，可以用在異步調(diào)用中，用來判斷作業(yè)完成或者取消作業(yè)，而后者則在內(nèi)部調(diào)用submitJob，阻塞等待直到作業(yè)完成（或失敗）。4.3 作業(yè)調(diào)度模塊頂層邏輯概述 DAGScheduler在Spar

4、kContext初始化的過程中被實(shí)例化，一個(gè)SparkContext對(duì)應(yīng)創(chuàng)建一個(gè)DAGScheduler。 DAGScheduler在初始化過程中會(huì)創(chuàng)建一個(gè)DAGSchedulerEventProcessActor實(shí)例來處理各種DAGSchedulerEvent事件，這些事件包括作業(yè)的提交、任務(wù)狀態(tài)的變化、監(jiān)控等，如程序4-2所示，DAGSchedulerEventProcessActor的Receive函數(shù)是DAGScheduler中處理這些事件的入口，可以看到各種待處理的事件。4.3 作業(yè)調(diào)度模塊頂層邏輯概述4.3 作業(yè)調(diào)度模塊頂層邏輯概述 eventProcessActor是DAGSch

5、eduler的私有成員，通過DAGScheduler封裝好的公共函數(shù)接口間接地向eventProcessActor發(fā)送相關(guān)消息，如在前面提到的submitJob，如程序4-3所示。4.3 作業(yè)調(diào)度模塊頂層邏輯概述 DAGScheduler內(nèi)部維護(hù)了各種“任務(wù)/調(diào)度階段/作業(yè)”的狀態(tài)和相互之間的映射關(guān)系表，用于在任務(wù)狀態(tài)更新、集群狀態(tài)更新等各種情況下，正確的維護(hù)作業(yè)的運(yùn)行邏輯。4.4 作業(yè)調(diào)度具體工作流程 每個(gè)作業(yè)從提交到完成，都要經(jīng)歷多個(gè)步驟，拆分成以任務(wù)為最小單位，按照一定的邏輯依賴關(guān)系依次提交執(zhí)行，并返回結(jié)果。4.4 作業(yè)調(diào)度具體工作流程4.4.1 調(diào)度階段的拆分4.4.1 調(diào)度階段的拆分

6、4.4.1 調(diào)度階段的拆分 以GroupByKey操作為例，該操作返回的結(jié)果實(shí)際上是一個(gè)ShuffleRDD，當(dāng)DAGScheduler遍歷到這個(gè)ShuffleRDD的時(shí)候，因?yàn)槠湟蕾囮P(guān)系是一個(gè)ShuffleDependency，于是這個(gè)ShuffleRDD的父RDD以及ShuffleDependency等對(duì)象就被用來構(gòu)建一個(gè)新的Stage，這個(gè)Stage的輸出結(jié)果的分區(qū)方式，則由ShuffleDependency 中的Partitioner對(duì)象決定。 盡管劃分和構(gòu)建Stage的依據(jù)是ShuffleDependency，在前面的例子中對(duì)應(yīng)的RDD是ShuffleRDD，但是這個(gè)Stage所處理

7、的數(shù)據(jù)是從這個(gè)ShuffleRDD的父RDD開始的，只是最終的輸出結(jié)果位置信息參考了ShuffleRDD返回的ShuffleDependency里所包含的內(nèi)容。而ShuffleRDD本身的運(yùn)算操作，是在下一個(gè)Stage里進(jìn)行的。下面以pagerank的程序4-5為例說明。4.4.1 調(diào)度階段的拆分4.4.1 調(diào)度階段的拆分4.4.2 調(diào)度階段的提交4.4.2 調(diào)度階段的提交 將重新掃描一次等待列表中的所有調(diào)度階段的列表，檢查它們是否還有任何依賴的調(diào)度階段沒有完成，如果沒有，說明該調(diào)度階段處于就緒狀態(tài)，就可以再次嘗試提交該階段了。4.4.3 任務(wù)集的提交調(diào)度階段的提交，最終會(huì)被轉(zhuǎn)換成一個(gè)任務(wù)集的

8、提交，DAGScheduler通過TaskScheduler接口提交任務(wù)集，這個(gè)任務(wù)集最終會(huì)觸發(fā)TaskScheduler構(gòu)建一個(gè)TaskSetManager的實(shí)例來管理這個(gè)任務(wù)集的生命周期，對(duì)于DAGScheduler來說，提交調(diào)度階段的工作到此就完成了。而TaskScheduler的具體實(shí)現(xiàn)則會(huì)在得到計(jì)算資源的時(shí)候，進(jìn)一步通過TaskSetManager調(diào)度具體的任務(wù)到對(duì)應(yīng)的Executor節(jié)點(diǎn)上進(jìn)行運(yùn)算。4.4.4 完成狀態(tài)的監(jiān)控 要保證相互依賴的作業(yè)調(diào)度階段能夠得到順利的調(diào)度執(zhí)行，DAGScheduler需要監(jiān)控當(dāng)前作業(yè)調(diào)度階段乃至任務(wù)的完成情況。這通過對(duì)外暴露一系列的回調(diào)函數(shù)來實(shí)現(xiàn)的

9、，對(duì)于TaskScheduler來說，這些回調(diào)函數(shù)主要包括任務(wù)的開始結(jié)束失敗、任務(wù)集的失敗，DAGScheduler根據(jù)這些任務(wù)的生命周期信息進(jìn)一步維護(hù)作業(yè)和調(diào)度階段的狀態(tài)信息。 TaskScheduler還可以通過回調(diào)函數(shù)通知DAGScheduler具體的Executor的生命狀態(tài)，如果某一個(gè)Executor崩潰了，則對(duì)應(yīng)的調(diào)度階段任務(wù)集的ShuffleMapTask的輸出結(jié)果也將標(biāo)志為不可用，這將導(dǎo)致對(duì)應(yīng)任務(wù)集狀態(tài)的變更，進(jìn)而重新執(zhí)行相關(guān)計(jì)算任務(wù)，以獲取丟失的相關(guān)數(shù)據(jù)。4.4.5 任務(wù)結(jié)果的獲取一個(gè)具體的任務(wù)在Executor中執(zhí)行完畢后，其結(jié)果需要以某種形式返回給DAGScheduler

10、，根據(jù)任務(wù)類型的不同，任務(wù)結(jié)果的返回方式也不同。對(duì)于FinalStage所對(duì)應(yīng)的任務(wù)，返回給DAGScheduler的是運(yùn)算結(jié)果本身，而對(duì)于中間調(diào)度階段對(duì)應(yīng)的任務(wù)ShuffleMapTask，返回給DAGScheduler的是一個(gè)MapStatus里的相關(guān)存儲(chǔ)信息，而非結(jié)果本身，這些存儲(chǔ)位置信息將作為下一個(gè)調(diào)度階段的任務(wù)獲取輸入數(shù)據(jù)的依據(jù)。根據(jù)任務(wù)結(jié)果大小的不同，ResultTask返回的結(jié)果又分為兩類，如果結(jié)果足夠小，則直接放在DirectTaskResult對(duì)象內(nèi)中，如果超過特定尺寸則在Executor端會(huì)將DirectTaskResult先序列化，再把序列化的結(jié)果作為一個(gè)數(shù)據(jù)塊存放在Bl

11、ockManager中，然后將BlockManager返回的BlockID放在IndirectTaskResult對(duì)象中返回給TaskScheduler，TaskScheduler進(jìn)而調(diào)用TaskResultGetter將IndirectTaskResult中的BlockID取出并通過BlockManager最終取得對(duì)應(yīng)的DirectTaskResult。4.5 任務(wù)集管理模塊TaskSetManager的主要接口：ResourceOffer：根據(jù)TaskScheduler所提供的單個(gè)Resource資源以及任務(wù)的host、executor和locality本地性的要求返回一個(gè)合適的任務(wù)。handleSuccessfulTask/handleFailedTask/handleTaskGettingResult：用于更新任務(wù)的運(yùn)行狀態(tài)，TaskSetManager在這些函數(shù)中除了更新自身維護(hù)的任務(wù)狀態(tài)列表等信息以用于剩余的任務(wù)調(diào)度外，還會(huì)進(jìn)一步調(diào)用DAGScheduler的函數(shù)接口將結(jié)果通知給它。4.6 調(diào)度池和調(diào)度模式分析每個(gè)SparkContext可能又同時(shí)存在多個(gè)可運(yùn)行的任務(wù)集，這些任務(wù)集之間如何調(diào)度，則是