在大數(shù)據(jù)領(lǐng)域，只有深挖數(shù)據(jù)科學(xué)領(lǐng)域，走在學(xué)術(shù)前沿，才能在底層算法和模型方面走在前面，從而占據(jù)領(lǐng)先地位。

Spark的這種學(xué)術(shù)基因，使得它從一開(kāi)始就在大數(shù)據(jù)領(lǐng)域建立了一定優(yōu)勢(shì)。無(wú)論是性能，還是方案的統(tǒng)一性，對(duì)比傳統(tǒng)的Hadoop，優(yōu)勢(shì)都非常明顯。Spark提供的基于RDD的一體化解決方案，將MapReduce、Streaming、SQL、Machine Learning、Graph Processing等模型統(tǒng)一到一個(gè)平臺(tái)下，并以一致的API公開(kāi)，并提供相同的部署方案，使得Spark的工程應(yīng)用領(lǐng)域變得更加廣泛。本文主要分以下章節(jié)：

• 一、Spark專業(yè)術(shù)語(yǔ)定義
• 二、Spark運(yùn)行基本流程
• 三、Spark運(yùn)行架構(gòu)特點(diǎn)
• 四、Spark核心原理透視

一、Spark專業(yè)術(shù)語(yǔ)定義

1、Application：Spark應(yīng)用程序

指的是用戶編寫的Spark應(yīng)用程序，包含了Driver功能代碼和分布在集群中多個(gè)節(jié)點(diǎn)上運(yùn)行的Executor代碼。

Spark應(yīng)用程序，由一個(gè)或多個(gè)作業(yè)JOB組成，如下圖所示:

2、Driver：驅(qū)動(dòng)程序

Spark中的Driver即運(yùn)行上述Application的Main()函數(shù)并且創(chuàng)建SparkContext，其中創(chuàng)建SparkContext的目的是為了準(zhǔn)備Spark應(yīng)用程序的運(yùn)行環(huán)境。在Spark中由SparkContext負(fù)責(zé)和ClusterManager通信，進(jìn)行資源的申請(qǐng)、任務(wù)的分配和監(jiān)控等;當(dāng)Executor部分運(yùn)行完畢后，Driver負(fù)責(zé)將SparkContext關(guān)閉。通常SparkContext代表Driver，如下圖所示:

3、Cluster Manager：資源管理器

指的是在集群上獲取資源的外部服務(wù)，常用的有：Standalone，Spark原生的資源管理器，由Master負(fù)責(zé)資源的分配;Haddop Yarn，由Yarn中的ResearchManager負(fù)責(zé)資源的分配;Messos，由Messos中的Messos Master負(fù)責(zé)資源管理，如下圖所示:

4、Executor：執(zhí)行器

Application運(yùn)行在Worker節(jié)點(diǎn)上的一個(gè)進(jìn)程，該進(jìn)程負(fù)責(zé)運(yùn)行Task，并且負(fù)責(zé)將數(shù)據(jù)存在內(nèi)存或者磁盤上，每個(gè)Application都有各自獨(dú)立的一批Executor，如下圖所示:

5、Worker：計(jì)算節(jié)點(diǎn)

集群中任何可以運(yùn)行Application代碼的節(jié)點(diǎn)，類似于Yarn中的NodeManager節(jié)點(diǎn)。在Standalone模式中指的就是通過(guò)Slave文件配置的Worker節(jié)點(diǎn)，在Spark on Yarn模式中指的就是NodeManager節(jié)點(diǎn)，在Spark on Messos模式中指的就是Messos Slave節(jié)點(diǎn)，如下圖所示:

6、RDD：彈性分布式數(shù)據(jù)集

Resillient Distributed Dataset，Spark的基本計(jì)算單元，可以通過(guò)一系列算子進(jìn)行操作(主要有Transformation和Action操作)，如下圖所示:

7、窄依賴

父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ū)。如圖所示:

8、寬依賴

父RDD的每個(gè)分區(qū)都可能被多個(gè)子RDD分區(qū)所使用，子RDD分區(qū)通常對(duì)應(yīng)所有的父RDD分區(qū)。如圖所示:

常見(jiàn)的窄依賴有：map、filter、union、mapPartitions、mapValues、join(父RDD是hash-partitioned ：如果JoinAPI之前被調(diào)用的RDD API是寬依賴(存在shuffle), 而且兩個(gè)join的RDD的分區(qū)數(shù)量一致，join結(jié)果的rdd分區(qū)數(shù)量也一樣，這個(gè)時(shí)候join api是窄依賴)。

常見(jiàn)的寬依賴有g(shù)roupByKey、partitionBy、reduceByKey、join(父RDD不是hash-partitioned ：除此之外的，rdd 的join api是寬依賴)。

9、DAG：有向無(wú)環(huán)圖

Directed Acycle graph，反應(yīng)RDD之間的依賴關(guān)系，如圖所示:

10、DAGScheduler：有向無(wú)環(huán)圖調(diào)度器

基于DAG劃分Stage 并以TaskSet的形勢(shì)提交Stage給TaskScheduler;負(fù)責(zé)將作業(yè)拆分成不同階段的具有依賴關(guān)系的多批任務(wù);最重要的任務(wù)之一就是：計(jì)算作業(yè)和任務(wù)的依賴關(guān)系，制定調(diào)度邏輯。在SparkContext初始化的過(guò)程中被實(shí)例化，一個(gè)SparkContext對(duì)應(yīng)創(chuàng)建一個(gè)DAGScheduler。

11、TaskScheduler：任務(wù)調(diào)度器

將Taskset提交給worker(集群)運(yùn)行并回報(bào)結(jié)果;負(fù)責(zé)每個(gè)具體任務(wù)的實(shí)際物理調(diào)度。如圖所示:

12、Job：作業(yè)

由一個(gè)或多個(gè)調(diào)度階段所組成的一次計(jì)算作業(yè);包含多個(gè)Task組成的并行計(jì)算，往往由Spark Action催生，一個(gè)JOB包含多個(gè)RDD及作用于相應(yīng)RDD上的各種Operation。如圖所示:

13、Stage：調(diào)度階段

一個(gè)任務(wù)集對(duì)應(yīng)的調(diào)度階段;每個(gè)Job會(huì)被拆分很多組Task，每組任務(wù)被稱為Stage，也可稱TaskSet，一個(gè)作業(yè)分為多個(gè)階段;Stage分成兩種類型ShuffleMapStage、ResultStage。如圖所示:

14、TaskSet：任務(wù)集

由一組關(guān)聯(lián)的，但相互之間沒(méi)有Shuffle依賴關(guān)系的任務(wù)所組成的任務(wù)集。如圖所示:

提示：

• 一個(gè)Stage創(chuàng)建一個(gè)TaskSet;
• 為Stage的每個(gè)Rdd分區(qū)創(chuàng)建一個(gè)Task,多個(gè)Task封裝成TaskSet

15、Task：任務(wù)

被送到某個(gè)Executor上的工作任務(wù);單個(gè)分區(qū)數(shù)據(jù)集上的最小處理流程單元。如圖所示:

總體如圖所示：

二、Spark運(yùn)行基本流程

三、Spark運(yùn)行架構(gòu)特點(diǎn)

1、Executor進(jìn)程專屬

每個(gè)Application獲取專屬的executor進(jìn)程，該進(jìn)程在Application期間一直駐留，并以多線程方式運(yùn)行tasks。Spark Application不能跨應(yīng)用程序共享數(shù)據(jù)，除非將數(shù)據(jù)寫入到外部存儲(chǔ)系統(tǒng)。如圖所示:

2、支持多種資源管理器

Spark與資源管理器無(wú)關(guān)，只要能夠獲取executor進(jìn)程，并能保持相互通信就可以了，Spark支持資源管理器包含： Standalone、On Mesos、On YARN、Or On EC2。如圖所示:

3、Job提交就近原則

提交SparkContext的Client應(yīng)該靠近Worker節(jié)點(diǎn)(運(yùn)行Executor的節(jié)點(diǎn))，***是在同一個(gè)Rack(機(jī)架)里，因?yàn)镾park Application運(yùn)行過(guò)程中SparkContext和Executor之間有大量的信息交換;如果想在遠(yuǎn)程集群中運(yùn)行，***使用RPC將SparkContext提交給集群，不要遠(yuǎn)離Worker運(yùn)行SparkContext。如圖所示:

4、移動(dòng)程序而非移動(dòng)數(shù)據(jù)的原則執(zhí)行

Task采用了數(shù)據(jù)本地性和推測(cè)執(zhí)行的優(yōu)化機(jī)制。關(guān)鍵方法：taskIdToLocations、getPreferedLocations。如圖所示:

四、Spark核心原理透視

1、計(jì)算流程

2、從代碼構(gòu)建DAG圖

Spark program

Val lines1 = sc.textFile(inputPath1). map(···)). map(···)  
Val lines2 = sc.textFile(inputPath2) . map(···)  
Val lines3 = sc.textFile(inputPath3)  
Val dtinone1 = lines2.union(lines3)  
Val dtinone = lines1.join(dtinone1)  
dtinone.saveAsTextFile(···)  
dtinone.filter(···).foreach(···) 

Spark的計(jì)算發(fā)生在RDD的Action操作，而對(duì)Action之前的所有Transformation，Spark只是記錄下RDD生成的軌跡，而不會(huì)觸發(fā)真正的計(jì)算。

Spark內(nèi)核會(huì)在需要計(jì)算發(fā)生的時(shí)刻繪制一張關(guān)于計(jì)算路徑的有向無(wú)環(huán)圖，也就是DAG。

3、將DAG劃分為Stage核心算法

Application多個(gè)job多個(gè)Stage：Spark Application中可以因?yàn)椴煌腁ction觸發(fā)眾多的job，一個(gè)Application中可以有很多的job，每個(gè)job是由一個(gè)或者多個(gè)Stage構(gòu)成的，后面的Stage依賴于前面的Stage，也就是說(shuō)只有前面依賴的Stage計(jì)算完畢后，后面的Stage才會(huì)運(yùn)行。

劃分依據(jù)：Stage劃分的依據(jù)就是寬依賴，何時(shí)產(chǎn)生寬依賴，reduceByKey, groupByKey等算子，會(huì)導(dǎo)致寬依賴的產(chǎn)生。

核心算法：從后往前回溯，遇到窄依賴加入本stage，遇見(jiàn)寬依賴進(jìn)行Stage切分。Spark內(nèi)核會(huì)從觸發(fā)Action操作的那個(gè)RDD開(kāi)始從后往前推，首先會(huì)為***一個(gè)RDD創(chuàng)建一個(gè)stage，然后繼續(xù)倒推，如果發(fā)現(xiàn)對(duì)某個(gè)RDD是寬依賴，那么就會(huì)將寬依賴的那個(gè)RDD創(chuàng)建一個(gè)新的stage，那個(gè)RDD就是新的stage的***一個(gè)RDD。然后依次類推，繼續(xù)繼續(xù)倒推，根據(jù)窄依賴或者寬依賴進(jìn)行stage的劃分，直到所有的RDD全部遍歷完成為止。

4、將DAG劃分為Stage剖析

從HDFS中讀入數(shù)據(jù)生成3個(gè)不同的RDD，通過(guò)一系列transformation操作后再將計(jì)算結(jié)果保存回HDFS。可以看到這個(gè)DAG中只有join操作是一個(gè)寬依賴，Spark內(nèi)核會(huì)以此為邊界將其前后劃分成不同的Stage. 同時(shí)我們可以注意到，在圖中Stage2中，從map到union都是窄依賴，這兩步操作可以形成一個(gè)流水線操作，通過(guò)map操作生成的partition可以不用等待整個(gè)RDD計(jì)算結(jié)束，而是繼續(xù)進(jìn)行union操作，這樣大大提高了計(jì)算的效率。

5、相關(guān)代碼

6、提交Stages

調(diào)度階段的提交，最終會(huì)被轉(zhuǎn)換成一個(gè)任務(wù)集的提交，DAGScheduler通過(guò)TaskScheduler接口提交任務(wù)集，這個(gè)任務(wù)集最終會(huì)觸發(fā)TaskScheduler構(gòu)建一個(gè)TaskSetManager的實(shí)例來(lái)管理這個(gè)任務(wù)集的生命周期，對(duì)于DAGScheduler來(lái)說(shuō)，提交調(diào)度階段的工作到此就完成了。而TaskScheduler的具體實(shí)現(xiàn)則會(huì)在得到計(jì)算資源的時(shí)候，進(jìn)一步通過(guò)TaskSetManager調(diào)度具體的任務(wù)到對(duì)應(yīng)的Executor節(jié)點(diǎn)上進(jìn)行運(yùn)算。

7、相關(guān)代碼

TaskSetManager負(fù)責(zé)管理TaskSchedulerImpl中一個(gè)單獨(dú)TaskSet，跟蹤每一個(gè)task，如果task失敗，負(fù)責(zé)重試task直到達(dá)到task重試次數(shù)的最多次數(shù)。

8、監(jiān)控Job、Task、Executor

DAGScheduler監(jiān)控Job與Task：要保證相互依賴的作業(yè)調(diào)度階段能夠得到順利的調(diào)度執(zhí)行，DAGScheduler需要監(jiān)控當(dāng)前作業(yè)調(diào)度階段乃至任務(wù)的完成情況。這通過(guò)對(duì)外暴露一系列的回調(diào)函數(shù)來(lái)實(shí)現(xiàn)的，對(duì)于TaskScheduler來(lái)說(shuō)，這些回調(diào)函數(shù)主要包括任務(wù)的開(kāi)始結(jié)束失敗、任務(wù)集的失敗，DAGScheduler根據(jù)這些任務(wù)的生命周期信息進(jìn)一步維護(hù)作業(yè)和調(diào)度階段的狀態(tài)信息。

DAGScheduler監(jiān)控Executor的生命狀態(tài)：TaskScheduler通過(guò)回調(diào)函數(shù)通知DAGScheduler具體的Executor的生命狀態(tài)，如果某一個(gè)Executor崩潰了，則對(duì)應(yīng)的調(diào)度階段任務(wù)集的ShuffleMapTask的輸出結(jié)果也將標(biāo)志為不可用，這將導(dǎo)致對(duì)應(yīng)任務(wù)集狀態(tài)的變更，進(jìn)而重新執(zhí)行相關(guān)計(jì)算任務(wù)，以獲取丟失的相關(guān)數(shù)據(jù)。

9、獲取任務(wù)執(zhí)行結(jié)果

結(jié)果DAGScheduler：一個(gè)具體的任務(wù)在Executor中執(zhí)行完畢后，其結(jié)果需要以某種形式返回給DAGScheduler，根據(jù)任務(wù)類型的不同，任務(wù)結(jié)果的返回方式也不同。

兩種結(jié)果，中間結(jié)果與最終結(jié)果：對(duì)于FinalStage所對(duì)應(yīng)的任務(wù)，返回給DAGScheduler的是運(yùn)算結(jié)果本身，而對(duì)于中間調(diào)度階段對(duì)應(yīng)的任務(wù)ShuffleMapTask，返回給DAGScheduler的是一個(gè)MapStatus里的相關(guān)存儲(chǔ)信息，而非結(jié)果本身，這些存儲(chǔ)位置信息將作為下一個(gè)調(diào)度階段的任務(wù)獲取輸入數(shù)據(jù)的依據(jù)。

兩種類型，DirectTaskResult與IndirectTaskResult：根據(jù)任務(wù)結(jié)果大小的不同，ResultTask返回的結(jié)果又分為兩類，如果結(jié)果足夠小，則直接放在DirectTaskResult對(duì)象內(nèi)中，如果超過(guò)特定尺寸則在Executor端會(huì)將DirectTaskResult先序列化，再把序列化的結(jié)果作為一個(gè)數(shù)據(jù)塊存放在BlockManager中，然后將BlockManager返回的BlockID放在IndirectTaskResult對(duì)象中返回給TaskScheduler，TaskScheduler進(jìn)而調(diào)用TaskResultGetter將IndirectTaskResult中的BlockID取出并通過(guò)BlockManager最終取得對(duì)應(yīng)的DirectTaskResult。

10、任務(wù)調(diào)度總體詮釋