前言：

由于最近對spark的運行流程非常感興趣，所以閱讀了《Spark大數(shù)據(jù)處理：技術(shù)、應(yīng)用與性能優(yōu)化》一書。通過這本書的學(xué)習(xí)，了解了spark的核心技術(shù)、實際應(yīng)用場景以及性能優(yōu)化的方法。本文旨在記錄和分享下spark運行的基本流程。

一、spark的基礎(chǔ)組件及其概念

1. ClusterManager

在Standalone模式中即為Master，控制整個集群，監(jiān)控Worker。在YARN模式中為資源管理器。

2. Application

用戶自定義的spark程序， 用戶提交后， Spark為App分配資源， 將程序轉(zhuǎn)換并執(zhí)行。

3. Driver

在Spark中，driver是一個核心概念，指的是Spark應(yīng)用程序的主進(jìn)程，也稱為主節(jié)點。負(fù)責(zé)運行Application的main（ ） 函數(shù)并創(chuàng)建SparkContext。

4. Worker

從節(jié)點，負(fù)責(zé)控制計算節(jié)點，啟動Executor或Driver。在YARN模式中為NodeManager，負(fù)責(zé)計算節(jié)點的控制。

5. Executor

執(zhí)行器，在Worker節(jié)點上執(zhí)行任務(wù)的組件、用于啟動線程池運行任務(wù)。每個Application擁有獨立的一組Executors。

6. RDD Graph

RDD是spark的核心結(jié)構(gòu)， 可以通過一系列算子進(jìn)行操作（ 主要有Transformation和Action操作） 。 當(dāng)RDD遇到Action算子時， 將之前的所有算子形成一個有向無環(huán)圖（ DAG） ， 也就是RDD Graph。 再在Spark中轉(zhuǎn)化為Job， 提交到集群執(zhí)行。一個App中可以包含多個Job。

7. Job

一個RDD Graph觸發(fā)的作業(yè)， 往往由Spark Action算子觸發(fā)， 在SparkContext中通過runJob方法向Spark提交Job。

8. Stage

每個Job會根據(jù)RDD的寬依賴關(guān)系被切分很多Stage， 每個Stage中包含一組相同的Task， 這一組Task也叫TaskSet。

9. Task

一個分區(qū)對應(yīng)一個Task， Task執(zhí)行RDD中對應(yīng)Stage中包含的算子。 Task被封裝好后放入Executor的線程池中執(zhí)行。

二、spark架構(gòu)

spark架構(gòu)采用了分布式計算中的Master-Slave模型。Master作為整個集群的控制器，負(fù)責(zé)整個集群的正常運行；Worker相當(dāng)于是計算節(jié)點，接收主節(jié)點命令與進(jìn)行狀態(tài)匯報；Executor負(fù)責(zé)任務(wù)的執(zhí)行；Client作為用戶的客戶端負(fù)責(zé)提交應(yīng)用，Driver負(fù)責(zé)控制一個應(yīng)用的執(zhí)行。

?

??

如圖所示，spark集群部署后，需要在主節(jié)點和從節(jié)點分別啟動Master進(jìn)程和Worker進(jìn)程，對整個集群進(jìn)行控制。在一個spark應(yīng)用的執(zhí)行過程中，Driver和Worker是兩個重要角色。Driver程序是應(yīng)用邏輯執(zhí)行的起點，負(fù)責(zé)作業(yè)的調(diào)度，即Task任務(wù)的分發(fā)，而多個Worker用來管理計算節(jié)點和創(chuàng)建Executor并行處理任務(wù)。在執(zhí)行階段，Driver會將Task和Task所依賴的file和jar序列化后傳遞給對應(yīng)的Worker機(jī)器，同時Executor對相應(yīng)數(shù)據(jù)分區(qū)的任務(wù)進(jìn)行處理。

三、Spark的工作機(jī)制

1. Spark的整體流程

Client提交應(yīng)用，Master找到一個Worker啟動Driver，Driver向Master或者資源管理器申請資源，之后將應(yīng)用轉(zhuǎn)化為RDD Graph，再由DAG Scheduler將RDD Graph轉(zhuǎn)化為Stage的有向無環(huán)圖提交給TaskScheduler，由TaskScheduler提交任務(wù)給Executor執(zhí)行。

?

??

如圖所示，在spark應(yīng)用中，整個執(zhí)行流程在邏輯上會形成有向無環(huán)圖。Action算子觸發(fā)之后，將所有累計的算子形成一個有向無環(huán)圖，然后由調(diào)度器調(diào)度該圖上的任務(wù)進(jìn)行運算。spark根據(jù)RDD之間不同的依賴關(guān)系切分形成不同的階段（stage），一個階段包含一系列函數(shù)執(zhí)行流水線。途中A、B、C、D、E、F、分別代表不同的RDD，RDD內(nèi)的方框代表分區(qū)。數(shù)據(jù)從HDFS輸入spark，形成RDD A和RDD C，RDD C上執(zhí)行map操作，轉(zhuǎn)換為RDD D，RDD B和RDD E執(zhí)行Join操作，轉(zhuǎn)換為F。而在B和E連接轉(zhuǎn)化為F的過程中又會執(zhí)行Shuffle，最后RDD F通過函數(shù)saveAsSequenceFile輸出并保存到HDFS中。

2. Stage的劃分

如上面這個運行流程所示，在 Apache Spark 中，一個作業(yè)（Job）通常會被劃分為多個階段（Stage），每個階段包含一組并行的任務(wù)（Task）。這種劃分主要是基于數(shù)據(jù)寬窄依賴進(jìn)行的，以便更有效地進(jìn)行任務(wù)調(diào)度和執(zhí)行。以下是關(guān)于 Spark 中 Stage 劃分的一些關(guān)鍵點：

?寬窄依賴

窄依賴（Narrow Dependency）：父 RDD 的每個分區(qū)只會被一個子 RDD 的分區(qū)使用，或者多個子 RDD 分區(qū)計算時都使用同一個父 RDD 分區(qū)。窄依賴允許在一個集群節(jié)點上以流水線的方式（pipeline）計算所有父分區(qū)，不會造成網(wǎng)絡(luò)之間的數(shù)據(jù)混洗。

寬依賴（Wide Dependency）：父 RDD 的每個分區(qū)都可能被多個子 RDD 分區(qū)所使用，會引起 shuffle。

?Stage的劃分

Spark 根據(jù) RDD 之間的寬窄依賴關(guān)系來劃分 Stage。遇到寬依賴就劃分一個 Stage，每個 Stage 里面包含多個 Task，Task 的數(shù)量由該 Stage 最后一個 RDD 的分區(qū)數(shù)決定。一個 Stage 內(nèi)部的多個 Task 可以并行執(zhí)行，而 Stage 之間是串行執(zhí)行的。只有當(dāng)一個 Stage 中的所有 Task 都計算完成后，才會開始下一個 Stage 的計算。

?Shuffle 與 Stage 邊界

當(dāng) Spark 遇到一個寬依賴（如 `reduceByKey`、`groupBy` 等操作）時，它需要在該操作之前和之后分別創(chuàng)建一個新的 Stage。這是因為寬依賴需要 shuffle 數(shù)據(jù)，而 shuffle 通常涉及磁盤 I/O，因此將寬依賴作為 Stage 之間的邊界可以提高效率。

3. Stage和Task調(diào)度方式

Stage的調(diào)度是由DAGScheduler完成的。 由RDD的有向無環(huán)圖DAG切分出了Stage的有向無環(huán)圖DAG。 Stage的DAG通過最后執(zhí)行Stage為根進(jìn)行廣度優(yōu)先遍歷， 遍歷到最開始執(zhí)行的Stage執(zhí)行， 如果提交的Stage仍有未完成的父母Stage， 則Stage需要等待其父Stage執(zhí)行完才能執(zhí)行。 同時DAGScheduler中還維持了幾個重要的Key-Value集合構(gòu)， 用來記錄Stage的狀態(tài)， 這樣能夠避免過早執(zhí)行和重復(fù)提交Stage。waitingStages中記錄仍有未執(zhí)行的父母Stage， 防止過早執(zhí)行。 runningStages中保存正在執(zhí)行的Stage， 防止重復(fù)執(zhí)行。failedStages中保存執(zhí)行失敗的Stage， 需要重新執(zhí)行。

每個Stage包含一組并行的Task，這些Task被組織成TaskSet（任務(wù)集合）。DAGScheduler將劃分好的TaskSet提交給TaskScheduler。TaskScheduler是負(fù)責(zé)Task調(diào)度和集群資源管理的組件。TaskScheduler通過TaskSetManager來管理每個TaskSet。TaskSetManager會跟蹤和控制其管轄的Task的執(zhí)行，包括任務(wù)的啟動、狀態(tài)監(jiān)控和失敗重試等。當(dāng)TaskSet被提交到TaskScheduler時，TaskScheduler會決定在哪些Executor上運行Task，并通過集群管理器（如YARN、Mesos或Spark Standalone）將Task分發(fā)到相應(yīng)的節(jié)點上執(zhí)行。Executor接收到Task后，會在其管理的線程池中執(zhí)行任務(wù)。執(zhí)行過程中，Task的狀態(tài)會不斷更新，并通過狀態(tài)更新機(jī)制通知TaskSetManager。TaskSetManager根據(jù)接收到的狀態(tài)更新來跟蹤Task的執(zhí)行情況，如遇到任務(wù)失敗，會觸發(fā)重試機(jī)制直至達(dá)到設(shè)定的重試次數(shù)。

當(dāng)所有Task都執(zhí)行完成后，TaskScheduler會通知DAGScheduler，并由DAGScheduler負(fù)責(zé)觸發(fā)后續(xù)Stage的執(zhí)行（如果存在）。

4. Shuffle機(jī)制

為什么spark計算模型需要Shuffle過程？ 我們都知道， spark計算模型是在分布式的環(huán)境下計算的， 這就不可能在單進(jìn)程空間中容納所有的計算數(shù)據(jù)來進(jìn)行計算， 這樣數(shù)據(jù)就按照Key進(jìn)行分區(qū)， 分配成一塊一塊的小分區(qū)， 打散分布在集群的各個進(jìn)程的內(nèi)存空間中， 并不是所有計算算子都滿足于按照一種方式分區(qū)進(jìn)行計算。 例如， 當(dāng)需要對數(shù)據(jù)進(jìn)行排序存儲時， 就有了重新按照一定的規(guī)則對數(shù)據(jù)重新分區(qū)的必要， Shuffle就是包裹在各種需要重分區(qū)的算子之下的一個對數(shù)據(jù)進(jìn)行重新組合的過程。

?

?如圖， 整個Job分為Stage1~Stage3， 3個Stage。首先從最上端的Stage2、 Stage3執(zhí)行， 每個Stage對每個分區(qū)執(zhí)行變換（ transformation） 的流水線式的函數(shù)操作， 執(zhí)行到每個Stage最后階段進(jìn)行Shuffle Write，將數(shù)據(jù)重新根據(jù)下一個Stage分區(qū)數(shù)分成相應(yīng)的Bucket， 并將Bucket最后寫入磁盤。 這個過程就是Shuffle Write階段。執(zhí)行完Stage2、 Stage3之后， Stage1去存儲有Shuffle數(shù)據(jù)節(jié)點的磁盤Fetch需要的數(shù)據(jù)， 將數(shù)據(jù)Fetch到本地后進(jìn)行用戶定義的聚集函數(shù)操作。 這個階段叫Shuffle Fetch， Shuffle Fetch包含聚集階段。 這樣一輪一輪的Stage之間就完成了Shuffle操作。

四、結(jié)語

在閱讀《Spark大數(shù)據(jù)處理：技術(shù)、應(yīng)用與性能優(yōu)化》一書后，我大概了解了spark的運行機(jī)制及原理。上文僅是做了一個簡單的總結(jié)，而且并沒有對一些細(xì)節(jié)進(jìn)行深入解讀。在原書中有著十分詳細(xì)的介紹，包含其容錯、IO、網(wǎng)絡(luò)等機(jī)制以及從源碼解析spark的運行流程，而且書中通過大量實際案例，展示了如何在具體應(yīng)用中使用Spark進(jìn)行數(shù)據(jù)處理、分析和挖掘，使理論與實踐相結(jié)合，大家如有興趣可自行閱讀。

審核編輯 黃宇