Spark Stage

什么是Stage

Spark中的一個(gè)Stage只不過是物理執(zhí)行計(jì)劃其中的一個(gè)步驟，它是物理執(zhí)行計(jì)劃的一個(gè)執(zhí)行單元。一個(gè)Job會(huì)被拆分為多組Task，每組任務(wù)被稱為一個(gè)Stage，可以簡(jiǎn)單理解為MapReduce里面的Map Stage， Reduce Stage。

Spark的Job中Stage之間會(huì)有依賴關(guān)系?？梢岳眠@些依賴關(guān)系把Job所有stage串連起來形成一個(gè)有向無環(huán)圖。在單個(gè)job內(nèi)是根據(jù)shuffle算子來拆分stage的，shuffle之前是一個(gè)stage，shuffle之后是另一個(gè)stage，如果一個(gè)job中有多個(gè)shuffle，那么每個(gè)shuffle之前都是一個(gè)stage。一個(gè)job被提交運(yùn)行之后，將會(huì)觸發(fā)stage及其父stage的執(zhí)行。

窄依賴：
指父RDD的每一個(gè)分區(qū)最多被一個(gè)子RDD的分區(qū)所用，表現(xiàn)為一個(gè)父RDD的分區(qū)對(duì)應(yīng)于一個(gè)子RDD的分區(qū)，和兩個(gè)父RDD的分區(qū)對(duì)應(yīng)于一個(gè)子RDD 的分區(qū)。圖中，map/filter和union屬于第一類，對(duì)輸入進(jìn)行協(xié)同劃分（co-partitioned）的join屬于第二類。

寬依賴：
指子RDD的分區(qū)依賴于父RDD的所有分區(qū)，這是因?yàn)閟huffle類操作，如圖中的groupByKey和未經(jīng)協(xié)同劃分的join。

Stage:
一個(gè)Job會(huì)被拆分為多組Task，每組任務(wù)被稱為一個(gè)Stage就像Map Stage， Reduce Stage。Stage的劃分，簡(jiǎn)單的說是以ShuffleMapStage 和ResultStage 這兩種類型來劃分。

在Spark中有兩類task，一類是shuffleMapTask，一類是resultTask，第一類task的輸出是shuffle所需數(shù)據(jù)，第二類task的輸出是result，stage的劃分也以此為依據(jù)，shuffle之前的所有變換是一個(gè)stage，shuffle之后的操作是另一個(gè)stage。

比如 rdd.parallize(1 to 10).foreach(println) 這個(gè)操作沒有shuffle，直接就輸出了，那么只有它的task是resultTask，stage也只有一個(gè)；

如果是rdd.map(x => (x, 1)).reduceByKey(_ + _).foreach(println), 這個(gè)job因?yàn)橛衦educe，所以有一個(gè)shuffle過程，那么reduceByKey之前的是一個(gè)stage，執(zhí)行shuffleMapTask，輸出shuffle所需的數(shù)據(jù)，reduceByKey到最后是一個(gè)stage，直接就輸出結(jié)果了。

如果job中有多次shuffle，那么每個(gè)shuffle之前都是一個(gè)stage。

Spark會(huì)根據(jù)RDD之間的依賴關(guān)系將DAG圖劃分為不同的階段，對(duì)于窄依賴，由于partition依賴關(guān)系的確定性，partition的轉(zhuǎn)換處理就可以在同一個(gè)線程里完成，窄依賴就被spark劃分到同一個(gè)stage中，而對(duì)于寬依賴，只能等父RDD shuffle處理完成后，下一個(gè)stage才能開始接下來的計(jì)算。之所以稱之為ShuffleMapTask是因?yàn)樗枰獙⒆约旱挠?jì)算結(jié)果通過shuffle到下一個(gè)stage中。

Spark Stage的分類

在Spark中，Stage可以分成兩種類型。分別是：

• ShuffleMapStage

1. 這種Stage是以Shuffle為輸出邊界
2. 其輸入邊界可以是從外部獲取數(shù)據(jù)，也可以是另一個(gè)ShuffleMapStage的輸出
3. 其輸出可以是另一個(gè)Stage的開始
4. ShuffleMapStage的最后Task就是ShuffleMapTask
5. 在一個(gè)Job里可能有該類型的Stage，也可以能沒有該類型Stage

• ResultStage

1. 這種Stage是直接輸出結(jié)果
2. 其輸入邊界可以是從外部獲取數(shù)據(jù)，也可以是另一個(gè)ShuffleMapStage的輸出
3. ResultStage的最后Task就是ResultTask
4. 在一個(gè)Job里必定有該類型Stage

Stage類的定義

Stage類是一個(gè)抽象類，類的定義如下：

abstract class Stage {    def findMissingPartitions(): Seq[Int]}