大數據技術，Spark核心技術之運行原理

在大數據領域，只有深挖數據科學領域，走在學術前沿，才能在底層算法和模型方面走在前面，從而占據領先地位。

Spark的這種學術基因，使得它從一開始就在大數據領域建立了一定優勢。無論是性能，還是方案的統一性，對比傳統的Hadoop，優勢都非常明顯。Spark提供的基于RDD的一體化解決方案，將MapReduce、Streaming、SQL、Machine Learning、Graph Processing等模型統一到一個平臺下，并以一致的API公開，并提供相同的部署方案，使得Spark的工程應用領域變得更加廣泛。本文主要分以下章節：

一、Spark專業術語定義

二、Spark運行基本流程

三、Spark運行架構特點

四、Spark核心原理透視

一、Spark專業術語定義

1、Application：Spark應用程序

指的是用戶編寫的Spark應用程序，包含了Driver功能代碼和分布在集群中多個節點上運行的Executor代碼。

互聯網科技發展蓬勃興起，人工智能時代來臨，抓住下一個風口。為幫助那些往想互聯網方向轉行想學習，卻因為時間不夠，資源不足而放棄的人。我自己整理的一份最新的大數據進階資料和高級開發教程，大數據學習群：868847735? ?歡迎進階中和進想深入大數據的小伙伴加入。

Spark應用程序，由一個或多個作業JOB組成，如下圖所示:

3、Cluster Manager：資源管理器

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

5、Worker：計算節點

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

7、窄依賴

父RDD每一個分區最多被一個子RDD的分區所用；表現為一個父RDD的分區對應于一個子RDD的分區，或兩個父RDD的分區對應于一個子RDD 的分區。如圖所示:

常見的窄依賴有：map、filter、union、mapPartitions、mapValues、join（父RDD是hash-partitioned ：如果JoinAPI之前被調用的RDD API是寬依賴(存在shuffle), 而且兩個join的RDD的分區數量一致，join結果的rdd分區數量也一樣，這個時候join api是窄依賴）。

常見的寬依賴有groupByKey、partitionBy、reduceByKey、join（父RDD不是hash-partitioned ：除此之外的，rdd 的join api是寬依賴）。

9、DAG：有向無環圖

Directed Acycle graph，反應RDD之間的依賴關系，如圖所示:

11、TaskScheduler：任務調度器

將Taskset提交給worker（集群）運行并回報結果；負責每個具體任務的實際物理調度。如圖所示:

13、Stage：調度階段

一個任務集對應的調度階段；每個Job會被拆分很多組Task，每組任務被稱為Stage，也可稱TaskSet，一個作業分為多個階段；Stage分成兩種類型ShuffleMapStage、ResultStage。如圖所示:

提示：

1）一個Stage創建一個TaskSet；

2）為Stage的每個Rdd分區創建一個Task,多個Task封裝成TaskSet

15、Task：任務

被送到某個Executor上的工作任務；單個分區數據集上的最小處理流程單元。如圖所示:

二、Spark運行基本流程

三、Spark運行架構特點

1、Executor進程專屬

每個Application獲取專屬的executor進程，該進程在Application期間一直駐留，并以多線程方式運行tasks。Spark Application不能跨應用程序共享數據，除非將數據寫入到外部存儲系統。如圖所示:

3、Job提交就近原則

提交SparkContext的Client應該靠近Worker節點（運行Executor的節點)，最好是在同一個Rack（機架）里，因為Spark Application運行過程中SparkContext和Executor之間有大量的信息交換；如果想在遠程集群中運行，最好使用RPC將SparkContext提交給集群，不要遠離Worker運行SparkContext。如圖所示:

四、Spark核心原理透視

1、計算流程

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

Application多個job多個Stage：Spark Application中可以因為不同的Action觸發眾多的job，一個Application中可以有很多的job，每個job是由一個或者多個Stage構成的，后面的Stage依賴于前面的Stage，也就是說只有前面依賴的Stage計算完畢后，后面的Stage才會運行。

劃分依據：Stage劃分的依據就是寬依賴，何時產生寬依賴，reduceByKey, groupByKey等算子，會導致寬依賴的產生。

核心算法：從后往前回溯，遇到窄依賴加入本stage，遇見寬依賴進行Stage切分。Spark內核會從觸發Action操作的那個RDD開始從后往前推，首先會為最后一個RDD創建一個stage，然后繼續倒推，如果發現對某個RDD是寬依賴，那么就會將寬依賴的那個RDD創建一個新的stage，那個RDD就是新的stage的最后一個RDD。然后依次類推，繼續繼續倒推，根據窄依賴或者寬依賴進行stage的劃分，直到所有的RDD全部遍歷完成為止。

4、將DAG劃分為Stage剖析

從HDFS中讀入數據生成3個不同的RDD，通過一系列transformation操作后再將計算結果保存回HDFS?？梢钥吹竭@個DAG中只有join操作是一個寬依賴，Spark內核會以此為邊界將其前后劃分成不同的Stage. 同時我們可以注意到，在圖中Stage2中，從map到union都是窄依賴，這兩步操作可以形成一個流水線操作，通過map操作生成的partition可以不用等待整個RDD計算結束，而是繼續進行union操作，這樣大大提高了計算的效率。

7、相關代碼

8、監控Job、Task、Executor

DAGScheduler監控Job與Task：要保證相互依賴的作業調度階段能夠得到順利的調度執行，DAGScheduler需要監控當前作業調度階段乃至任務的完成情況。這通過對外暴露一系列的回調函數來實現的，對于TaskScheduler來說，這些回調函數主要包括任務的開始結束失敗、任務集的失敗，DAGScheduler根據這些任務的生命周期信息進一步維護作業和調度階段的狀態信息。

DAGScheduler監控Executor的生命狀態：TaskScheduler通過回調函數通知DAGScheduler具體的Executor的生命狀態，如果某一個Executor崩潰了，則對應的調度階段任務集的ShuffleMapTask的輸出結果也將標志為不可用，這將導致對應任務集狀態的變更，進而重新執行相關計算任務，以獲取丟失的相關數據。

9、獲取任務執行結果

結果DAGScheduler：一個具體的任務在Executor中執行完畢后，其結果需要以某種形式返回給DAGScheduler，根據任務類型的不同，任務結果的返回方式也不同。

兩種結果，中間結果與最終結果：對于FinalStage所對應的任務，返回給DAGScheduler的是運算結果本身，而對于中間調度階段對應的任務ShuffleMapTask，返回給DAGScheduler的是一個MapStatus里的相關存儲信息，而非結果本身，這些存儲位置信息將作為下一個調度階段的任務獲取輸入數據的依據。

兩種類型，DirectTaskResult與IndirectTaskResult：根據任務結果大小的不同，ResultTask返回的結果又分為兩類，如果結果足夠小，則直接放在DirectTaskResult對象內中，如果超過特定尺寸則在Executor端會將DirectTaskResult先序列化，再把序列化的結果作為一個數據塊存放在BlockManager中，然后將BlockManager返回的BlockID放在IndirectTaskResult對象中返回給TaskScheduler，TaskScheduler進而調用TaskResultGetter將IndirectTaskResult中的BlockID取出并通過BlockManager最終取得對應的DirectTaskResult。

10、任務調度總體詮釋

-----------------------------------------

本文轉自金羅老師博客51CTO

Spark 大數據

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。