spark運(yùn)行的基本流程

前言：

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

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

1. ClusterManager

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

2. Application

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

3. Driver

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

4. Worker

從節(jié)點(diǎn)，負(fù)責(zé)控制計(jì)算節(jié)點(diǎn)，啟動(dòng)Executor或Driver。在YARN模式中為NodeManager，負(fù)責(zé)計(jì)算節(jié)點(diǎn)的控制。

5. Executor

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

6. RDD Graph

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

7. Job

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

8. Stage

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

9. Task

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

二、spark架構(gòu)

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

?

??

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

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

1. Spark的整體流程

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

?

??

如圖所示，在spark應(yīng)用中，整個(gè)執(zhí)行流程在邏輯上會(huì)形成有向無(wú)環(huán)圖。Action算子觸發(fā)之后，將所有累計(jì)的算子形成一個(gè)有向無(wú)環(huán)圖，然后由調(diào)度器調(diào)度該圖上的任務(wù)進(jìn)行運(yùn)算。spark根據(jù)RDD之間不同的依賴關(guān)系切分形成不同的階段（stage），一個(gè)階段包含一系列函數(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的過程中又會(huì)執(zhí)行Shuffle，最后RDD F通過函數(shù)saveAsSequenceFile輸出并保存到HDFS中。

2. Stage的劃分

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

?寬窄依賴

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

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

?Stage的劃分

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

?Shuffle 與 Stage 邊界

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

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

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

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

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

4. Shuffle機(jī)制

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

?

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

四、結(jié)語(yǔ)

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

審核編輯 黃宇