一、Runtime架構圖

（1）從Spark  Runtime的角度講，包括五大核心物件：Master、Worker、Executor、Driver、CoarseGrainedExecutorBackend。

（2）Spark在做分散式集群系統設計的時候：最大化功能獨立、模組化封裝具體獨立的物件、強內聚鬆耦合。Spark執行架構圖如下圖所示。

（3）當Driver中的SparkContext初始化時會提交程式給Master，Master如果接受該程式在Spark中執行，就會為當前的程式分配AppID，同時會分配具體的計算資源。

需要特別注意的是，Master是根據當前提交程式的配置資訊來給叢集中的Worker發指令分配具體的計算資源，但是，Master發出指令後並不關心具體的資源是否已經分配，換言之，Master是發指令後就記錄了分配的資源，以後客戶端再次提交其他的程式，就不能使用該資源了。

其弊端是可能會導致其他要提交的程式無法分配到本來應該可以分配到的計算資源；最終的優勢是Spark分散式系統功能在耦合的基礎上最快的執行系統（否則如果Master要等到資源最終分配成功後才通知Driver，就會造成Driver阻塞，不能夠最大化平行計算資源的使用率）。

需要補充說明的是：Spark在預設情況下由於叢集中一般都只有一個Application在執行，所有Master分配資源策略的弊端就沒有那麼明顯了

二、生命週期

我們從Spark Runtime全域性的角度看Spark具體是怎麼工作的，從一個具體的job的視角通過Driver、Master、Worker、Executor等角色來具體看看Spark的Runtime生命週期

這裡我們編寫WordCountJobRuntime.scala程式碼，觀察日誌，源資料如下

程式碼如下：

import org.apache.log4j.{Level, Logger}

import org.apache.spark.{SparkConf, SparkContext}

 

object WordCountJobRuntime {

  def main(args: Array[String]){

    Logger.getLogger("org").setLevel(Level.ALL)

    //  第1步：建立Spark的配置物件SparkConf

    val conf = new SparkConf() //建立SparkConf物件

    conf.setAppName("Wow,WordCountJobRuntime!")

    //設定應用程式的名稱，在程式執行的監控介面可以看到名稱

    conf.setMaster("local") //此時，程式在本地執行，不需要安裝Spark叢集

    //第2步：建立SparkContext物件

    val sc = new SparkContext(conf)

 

    // 第 3  步：根據具體的資料來源（如  HDFS、HBase、Local FS、DB、S3等）通過 SparkContext建立RDD

    val lines = sc.textFile("WordCountJobRuntime.txt")

    //    第4步：對初始的RDD進行Transformation級別的處理，如通過map、filter等

    val words = lines.flatMap { line => line.split(" ")}

 

    val pairs = words.map { word => (word, 1) }

    val wordCountsOdered = pairs.reduceByKey(_+_)

    wordCountsOdered.foreach(println)

    while(true){

  

    }

    sc.stop()

 

    }

}

得到結果

在eclipse的控制檯中觀察WordCountJobRuntime.scala執行日誌，日誌中顯示FileInputFormat: Total input paths to process : 1說明有一個檔案要處理。

在Spark中，所有的Action都會觸發至少一個Job，在WordCountJobRuntime.scala程式碼中，是通過println來觸發Job的。緊接著交給DAGScheduler，日誌中顯示DAGScheduler: Registering RDD，因為這裡有兩個Stage，從具體計算的角度，前面Stage計算的時候保留輸出。然後是DAGScheduler獲得了job的ID（job 0）。

SparkContext在例項化的時候會構造StandaloneSchedulerBackend、DAGScheduler、TaskSchedulerImpl、MapOutputTrackerMaster等物件。

其中，StandaloneSchedulerBackend負責叢集計算資源的管理和排程，這是從作業的角度來考慮的，註冊給Master的時候，Master給我們分配資源，資源從Executor本身轉過來向StandaloneSchedulerBackend註冊，這是從作業排程的角度來考慮的，不是從整個叢集來考慮，整個叢集是Master來管理計算資源的。

DAGScheduler負責高層排程（如Job中Stage的劃分、資料本地性等內容）。

TaskSchedulerImple負責具體Stage內部的底層排程（如具體每個Task的排程、Task的容錯等）。

MapOutputTrackerMaster負責Shuffle中資料輸出和讀取的管理。Shuffle的時候將資料寫到本地，下一個Stage要使用上一個Stage的資料，因此寫資料的時候要告訴Driver中的MapOutputTrackerMaster具體寫到哪裡，下一個Stage讀取資料的時候也要訪問Driver的MapOutputTrackerMaster獲取資料的具體位置。

MapOutputTrackerMaster的原始碼如下。

private[spark] class MapOutputTrackerMaster(conf: SparkConf,

    broadcastManager: BroadcastManager, isLocal: Boolean)

  extends MapOutputTracker(conf) {

DAGScheduler是面向Stage排程的高層排程實現。它為每一個Job計算DAG，跟蹤RDDS及Stage輸出結果進行物化，並找到一個最小的計劃去執行Job，然後提交stages中的TaskSets到底層排程器TaskScheduler提交叢集執行，TaskSet包含完全獨立的任務，基於叢集上已存在的資料執行（如從上一個Stage輸出的檔案），如果這個資料不可用，獲取資料可能會失敗。

Spark Stages根據RDD圖中Shuffle的邊界來建立，如果RDD的操作是窄依賴，如map()和filter()，在每個Stages中將一系列tasks組合成流水線執行。但是，如果是寬依賴，Shuffle依賴需要多個Stages（上一個Stage進行map輸出寫入檔案，下一個Stage讀取資料檔案），每個Stage依賴於其他的Stage，其中進行多個運算元操作。運算元操作在各種型別的RDDS（如MappedRDD、FilteredRDD）的RDD.compute()中實際執行。

在DAG階段，DAGScheduler根據當前快取狀態決定每個任務執行的位置，並將任務傳遞給底層的任務排程器TaskScheduler。此外，它處理Shuffle輸出檔案丟失的故障，在這種情況下，以前的Stage可能需要重新提交。Stage中不引起Shuffle檔案丟失的故障由任務排程器TaskScheduler處理，在取消整個Stage前，將重試幾次任務。

當瀏覽這個程式碼時，有幾個關鍵概念：

　　Jobs作業（表現為[ActiveJob]）作為頂級工作項提交給排程程式。當用戶呼叫一個action，如count()運算元，Job將通過submitJob進行提交。每個作業可能需要執行多個stages來構建中間資料。

　　Stages ([Stage])是一組任務的集合，在相同的RDD分割槽上，每個任務計算相同的功能，計算Jobs的中間結果。Stage根據Shuffle劃分邊界，我們必須等待前一階段Stage完成輸出。有兩種型別的Stage：[ResultStage]是執行action的最後一個Stage，[ShuffleMapStage]是Shuffle Stages通過map寫入輸出檔案中的。如果Jobs重用相同的RDDs，Stages可以跨越多個Jobs共享。

　　Tasks任務是單獨的工作單位，每個任務傳送到一個分散式節點。

　　快取跟蹤：DAGScheduler記錄哪些RDDS被快取，避免重複計算，以及記錄Shuffle map Stages已經生成的輸出檔案，避免在map端重新計算。

　　資料本地化：DAGScheduler基於RDDS的資料本地性、快取位置，或Shuffle資料在Stage中執行每一個任務的Task。

　　清理：當依賴於它們的執行作業完成時，所有資料結構將被清除，防止在長期執行的應用程式中記憶體洩漏。

為了從故障中恢復，同一個Stage可能需要執行多次，這被稱為重試“attempts”。如在上一個Stage中的輸出檔案丟失，TaskScheduler中將報告任務失敗，DAGScheduler通過檢測CompletionEvent與FetchFailed或ExecutorLost事件重新提交丟失的Stage。DAGScheduler將等待看是否有其他節點或任務失敗，然後在丟失計算任務的階段Stage中重新提交TaskSets。在這個過程中，可能須建立之前被清理的Stage。舊Stage的任務仍然可以執行，但必須在正確的Stage中接收事件並進行操作。

DAGScheduler.scala的原始碼如下。

private[spark]

class DAGScheduler(

    private[scheduler] val sc: SparkContext,

    private[scheduler] val taskScheduler: TaskScheduler,

    listenerBus: LiveListenerBus,

    mapOutputTracker: MapOutputTrackerMaster,

    blockManagerMaster: BlockManagerMaster,

    env: SparkEnv,

    clock: Clock = new SystemClock())

  extends Logging {

回到執行日誌，SparkContext在例項化的時候會構造StandaloneSchedulerBackend、DAGScheduler、TaskSchedulerImpl、MapOutputTrackerMaster四大核心物件，DAGScheduler獲得Job ID，日誌中顯示DAGScheduler: Final stage: ResultStage 1，Final stage是ResultStage；Parents of final stage是ShuffleMapStage，DAGScheduler是面向Stage的。日誌中顯示兩個Stage：Stage 1是Final stage，Stage 0是ShuffleMapStage。

接下來序號改變，執行時最左側從0開始，日誌中顯示DAGScheduler: missing: List(ShuffleMapStage 0)，父Stage是ShuffleMapStage，DAGScheduler排程時必須先計算父Stage，因此首先提交的是ShuffleMapStage 0，這裡RDD是MapPartitionsRDD，只有Stage中的最後一個運算元是真正有效的，Stage 0中的最後一個操作是map，因此生成了MapPartitionsRDD。Stage 0無父Stage，因此提交，提交時進行廣播等內容，然後提交作業。

我們從http://172.20.10.3:4040/jobs/的角度看一下，如下圖所示，Web UI中顯示生成兩個Stage：Stage 0、Stage 1。

日誌中顯示DAGScheduler: Submitting 1 missing tasks from ShuffleMapStage 0，DAGScheduler提交作業，顯示提交一個須計算的任務，ShuffleMapStage在本地執行是一個並行度，交給TaskSchedulerImpl執行。

這裡是一個並行度，提交底層的排程器TaskScheduler，TaskScheduler收到任務後，就釋出任務到叢集中執行，由TaskSetManager進行管理：日誌中顯示TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, executor driver, partition 0, PROCESS_LOCAL, 6012 bytes)，顯示具體執行的位置，及worker運行了哪些任務。這裡在本地只運行了一個任務。

然後是完成作業，日誌中顯示TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 327 ms on localhost (executor driver)，在本地機器上完成作業。當Stage的一個任務完成後，ShuffleMapStage就已完成。Task任務執行完後向DAGScheduler彙報，DAGScheduler檢視曾經提交了幾個Task，計算Task的數量如果等於Task的總數量，那Stage也就完成了。這個Stage完成以後，下一個Stage開始執行。

ShuffleMapStage完成後，將執行下一個Stage。日誌中顯示DAGScheduler: looking for newly runnable stages，這裡一共有兩個Stage，ShuffleMapStage執行完成，那只有一個ResultStage將執行。DAGScheduler又提交最後一個Stage的一個任務，預設並行度是繼承的。同樣，釋出任務給Executor進行計算。

Task任務執行完後向DAGScheduler彙報，DAGScheduler計算曾經提交了幾個Task，如果Task的數量等於Task的總數量，ResultStage也執行完成。然後進行相關的清理工作，兩個Stage（ShuffleMapStage、ResultStage）完成，Job也就完成。

下面看一下WebUI，ShuffleMapStage中的任務交給Executor，圖3-9中顯示了任務的相關資訊，如Shuffle的輸出等，第一個Stage肯定生成Shuffle的輸出，可以看一下最右側的Shuffle Write Size/Records。下圖中的Input Size/Records是從Hdfs中讀入的檔案資料。

接下來看一下第二個Stage。第二個Stage同樣顯示Executor的資訊，下圖最右側顯示Shuffle Read Size/Records。如果在分散式叢集執行，須遠端讀取資料，例如，原來是4個Executor計算，在第二個Stage中是兩個Executor計算，因此一部分資料是本地的，一部分是遠端的，或從遠端節點拉取資料。ResultStage最後要產生輸出，輸出到檔案儲存。

三、總結

我們通過對Spark Runtime（Driver、Master、Worker、Executor）內幕解密，從Spark Runtime全域性的角度看Spark具體是怎麼工作的，從一個作業的視角通過Driver、Master、Worker、Executor等角色來透視Spark的Runtime生命週期。

這裡所有截圖都是從本機截圖的，都是能夠進行實際驗證的，包括所有原始碼是基於Spark-2.0.1的。

&n