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

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

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

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

4、kContext初始化的過程中被實例化，一個SparkContext對應創(chuàng)建一個DAGScheduler。 DAGScheduler在初始化過程中會創(chuàng)建一個DAGSchedulerEventProcessActor實例來處理各種DAGSchedulerEvent事件，這些事件包括作業(yè)的提交、任務狀態(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)部維護了各種“任務/調(diào)度階段/作業(yè)”的狀態(tài)和相互之間的映射關(guān)系表，用于在任務狀態(tài)更新、集群狀態(tài)更新等各種情況下，正確的維護作業(yè)的運行邏輯。4.4 作業(yè)調(diào)度具體工作流程 每個作業(yè)從提交到完成，都要經(jīng)歷多個步驟，拆分成以任務為最小單位，按照一定的邏輯依賴關(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é)果實際上是一個ShuffleRDD，當DAGScheduler遍歷到這個ShuffleRDD的時候，因為其依賴關(guān)系是一個ShuffleDependency，于是這個ShuffleRDD的父RDD以及ShuffleDependency等對象就被用來構(gòu)建一個新的Stage，這個Stage的輸出結(jié)果的分區(qū)方式，則由ShuffleDependency 中的Partitioner對象決定。 盡管劃分和構(gòu)建Stage的依據(jù)是ShuffleDependency，在前面的例子中對應的RDD是ShuffleRDD，但是這個Stage所處理

7、的數(shù)據(jù)是從這個ShuffleRDD的父RDD開始的，只是最終的輸出結(jié)果位置信息參考了ShuffleRDD返回的ShuffleDependency里所包含的內(nèi)容。而ShuffleRDD本身的運算操作，是在下一個Stage里進行的。下面以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 任務集的提交調(diào)度階段的提交，最終會被轉(zhuǎn)換成一個任務集的

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

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

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

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