Spark基本架構及原理

 Hadoop 和 Spark 的關係

Spark 運算比 Hadoop 的 MapReduce 框架快的緣由是由於 Hadoop 在一次 MapReduce 運算以後,會將數據的運算結果從內存寫入到磁盤中,第二次 Mapredue 運算時在從磁盤中讀取數據,因此其瓶頸在2次運算間的多餘 IO 消耗. Spark 則是將數據一直緩存在內存中,直到計算獲得最後的結果,再將結果寫入到磁盤,因此屢次運算的狀況下, Spark 是比較快的. 其優化了迭代式工做負載

Hadoop的侷限	Spark的改進
抽象層次低，代碼編寫難以上手	經過使用RDD的統一抽象，實現數據處理邏輯的代碼很是簡潔
只提供了Map和Reduce兩個操做，欠缺表達力	經過RDD提供了不少轉換和動做，實現了不少基本操做，如Sort, Join等
一個Job只有Map和Reduce兩個階段，複雜的程序須要大量的Job來完成，且Job之間的依賴關係須要開發者自行管理	一個Job能夠包含RDD的多個轉換操做，在調度時能夠生成多個階段（Stage），並且若是多個map操做的RDD的分區不變，是能夠放在同一個Task中進行
處理邏輯隱藏在代碼細節中，缺少總體邏輯視圖	RDD的轉換支持流式API,提供處理邏輯的總體視圖
對迭代式數據處理性能比較差，Reduce與下一步Map之間的中間結果只能存放在HDFS中	經過內存緩存數據，可大大提升迭代式計算的性能，內存不足時能夠溢出到本地磁盤，而不是HDFS
ReduceTask須要等待全部MapTask都完成後才能夠開始	分區相同的轉換構成流水線放在一個Task中運行，分區不一樣的轉換須要Shuffle，被劃分到不一樣的Stage中，須要等待前面的Stage完成後才能夠開始
時延高，只適用Batch數據處理，對於交互式數據處理和實時數據處理的支持不夠	經過將流拆成小的batch提供Discretized Stream處理流數據

 

Spark 的主要特色還包括:

• • (1)提供 Cache 機制來支持須要反覆迭代計算或者屢次數據共享,減小數據讀取的 IO 開銷;
  • (2)提供了一套支持 DAG 圖的分佈式並行計算的編程框架,減小屢次計算之間中間結果寫到 Hdfs 的開銷;
  • (3)使用多線程池模型減小 Task 啓動開稍, shuffle 過程當中避免沒必要要的 sort 操做並減小磁盤 IO 操做。(Hadoop 的 Map 和 reduce 之間的 shuffle 須要 sort)

Spark 系統架構

明確相關術語

 

• Application: Appliction都是指用戶編寫的Spark應用程序，其中包括一個Driver功能的代碼和分佈在集羣中多個節點上運行的Executor代碼
• Driver:  Spark中的Driver即運行上述Application的main函數並建立SparkContext，建立SparkContext的目的是爲了準備Spark應用程序的運行環境，在Spark中有SparkContext負責與ClusterManager通訊，進行資源申請、任務的分配和監控等，當Executor部分運行完畢後，Driver同時負責將SparkContext關閉，一般用SparkContext表明Driver
• Executor:  某個Application運行在worker節點上的一個進程，  該進程負責運行某些Task， 而且負責將數據存到內存或磁盤上，每一個Application都有各自獨立的一批Executor， 在Spark on Yarn模式下，其進程名稱爲CoarseGrainedExecutor Backend。一個CoarseGrainedExecutor Backend有且僅有一個Executor對象， 負責將Task包裝成taskRunner,並從線程池中抽取一個空閒線程運行Task， 這個每個oarseGrainedExecutor Backend能並行運行Task的數量取決與分配給它的cpu個數
• Cluter Manager：指的是在集羣上獲取資源的外部服務。目前有三種類型

 

1. 1. 　Standalon : spark原生的資源管理，由Master負責資源的分配
   2. 　Apache Mesos:與hadoop MR兼容性良好的一種資源調度框架
   3. 　Hadoop Yarn: 主要是指Yarn中的ResourceManager

 

• Worker: 集羣中任何能夠運行Application代碼的節點，在Standalone模式中指的是經過slave文件配置的Worker節點，在Spark on Yarn模式下就是NoteManager節點
• Task: 被送到某個Executor上的工做單元，但hadoopMR中的MapTask和ReduceTask概念同樣，是運行Application的基本單位，多個Task組成一個Stage，而Task的調度和管理等是由TaskScheduler負責
• Job: 包含多個Task組成的並行計算，每每由Spark Action觸發生成， 一個Application中每每會產生多個Job
• Stage: 每一個Job會被拆分紅多組Task， 做爲一個TaskSet， 其名稱爲Stage，Stage的劃分和調度是有DAGScheduler來負責的，Stage有非最終的Stage（Shuffle Map Stage）和最終的Stage（Result Stage）兩種，Stage的邊界就是發生shuffle的地方
• DAGScheduler: 根據Job構建基於Stage的DAG（Directed Acyclic Graph有向無環圖)，並提交Stage給TASkScheduler。 其劃分Stage的依據是RDD之間的依賴的關係找出開銷最小的調度方法，以下圖
• TASKSedulter: 將TaskSET提交給worker運行，每一個Executor運行什麼Task就是在此處分配的. TaskScheduler維護全部TaskSet，當Executor向Driver發生心跳時，TaskScheduler會根據資源剩餘狀況分配相應的Task。另外TaskScheduler還維護着全部Task的運行標籤，重試失敗的Task。下圖展現了TaskScheduler的做用
• 在不一樣運行模式中任務調度器具體爲：

 

1. 1. 　　Spark on Standalone模式爲TaskScheduler
   2. 　　YARN-Client模式爲YarnClientClusterScheduler
   3. 　　YARN-Cluster模式爲YarnClusterScheduler

 

• 將這些術語串起來的運行層次圖以下：
• Job=多個stage，Stage=多個同種task, Task分爲ShuffleMapTask和ResultTask，Dependency分爲ShuffleDependency和NarrowDependency

 

　　整個 Spark 集羣中,分爲 Master 節點與 worker 節點,,其中 Master 節點負責將串行任務變成可並行執行的任務集Tasks, 同時還負責出錯問題處理等,而 Worker 節點負責執行任務. 
　　Driver 的功能是建立 SparkContext, 負責執行用戶寫的 Application 的 main 函數進程,Application 就是用戶寫的程序. 
　　不一樣的模式可能會將 Driver 調度到不一樣的節點上執行.集羣管理模式裏, local 通常用於本地調試. 
　　每一個 Worker 上存在一個或多個 Executor 進程,該對象擁有一個線程池,每一個線程負責一個 Task 任務的執行.根據 Executor 上 CPU-core 的數量,其每一個時間能夠並行多個 跟 core 同樣數量的 Task.Task 任務即爲具體執行的 Spark 程序的任務. 

• spark運行流程圖以下：

1. 構建Spark Application的運行環境，啓動SparkContext
2. SparkContext向資源管理器（能夠是Standalone，Mesos，Yarn）申請運行Executor資源，並啓動StandaloneExecutorbackend，
3. Executor向SparkContext申請Task
4. SparkContext將應用程序分發給Executor
5. SparkContext構建成DAG圖，將DAG圖分解成Stage、將Taskset發送給Task Scheduler，最後由Task Scheduler將Task發送給Executor運行
6. Task在Executor上運行，運行完釋放全部資源

     Spark運行特色：

1. 每一個Application獲取專屬的executor進程，該進程在Application期間一直駐留，並以多線程方式運行Task。這種Application隔離機制是有優點的，不管是從調度角度看（每一個Driver調度他本身的任務），仍是從運行角度看（來自不一樣Application的Task運行在不一樣JVM中），固然這樣意味着Spark Application不能跨應用程序共享數據，除非將數據寫入外部存儲系統
2. Spark與資源管理器無關，只要可以獲取executor進程，並能保持相互通訊就能夠了
3. 提交SparkContext的Client應該靠近Worker節點（運行Executor的節點），最好是在同一個Rack裏，由於Spark Application運行過程當中SparkContext和Executor之間有大量的信息交換
4. Task採用了數據本地性和推測執行的優化機制

Spark做業基本運行原理

 

 

詳細原理見上圖。

　　咱們使用spark-submit提交一個Spark做業以後，這個做業就會啓動一個對應的Driver進程。根據你使用的部署模式（deploy-mode）不一樣，Driver進程可能在本地啓動，也可能在集羣中某個工做節點上啓動。Driver進程自己會根據咱們設置的參數，佔有必定數量的內存和CPU core。而Driver進程要作的第一件事情，就是向集羣管理器(YARN或者其餘資源管理集羣）申請運行Spark做業須要使用的資源，這裏的資源指的就是Executor進程。YARN集羣管理器會根據咱們爲Spark做業設置的資源參數，在各個工做節點上，啓動必定數量的Executor進程，每一個Executor進程都佔有必定數量的內存和CPU core。

　　在申請到了做業執行所需的資源以後，Driver進程就會開始調度和執行咱們編寫的做業代碼了。Driver進程會將咱們編寫的Spark做業代碼分拆爲多個stage，每一個stage執行一部分代碼片斷，併爲每一個stage建立一批task，而後將這些task分配到各個Executor進程中執行。task是最小的計算單元，負責執行如出一轍的計算邏輯（也就是咱們本身編寫的某個代碼片斷），只是每一個task處理的數據不一樣而已。一個stage的全部task都執行完畢以後，會在各個節點本地的磁盤文件中寫入計算中間結果，而後Driver就會調度運行下一個stage。下一個stage的task的輸入數據就是上一個stage輸出的中間結果。如此循環往復，直到將咱們本身編寫的代碼邏輯所有執行完，而且計算完全部的數據，獲得咱們想要的結果爲止。

　　Spark是根據shuffle類算子來進行stage的劃分。若是咱們的代碼中執行了某個shuffle類算子（好比reduceByKey、join等），那麼就會在該算子處，劃分出一個stage界限來。能夠大體理解爲，shuffle算子執行以前的代碼會被劃分爲一個stage，shuffle算子執行以及以後的代碼會被劃分爲下一個stage。所以一個stage剛開始執行的時候，它的每一個task可能都會從上一個stage的task所在的節點，去經過網絡傳輸拉取須要本身處理的全部key，而後對拉取到的全部相同的key使用咱們本身編寫的算子函數執行聚合操做（好比reduceByKey()算子接收的函數）。這個過程就是shuffle。

　　當咱們在代碼中執行了cache/persist等持久化操做時，根據咱們選擇的持久化級別的不一樣，每一個task計算出來的數據也會保存到Executor進程的內存或者所在節點的磁盤文件中。

　　所以Executor的內存主要分爲三塊：第一塊是讓task執行咱們本身編寫的代碼時使用，默認是佔Executor總內存的20%；第二塊是讓task經過shuffle過程拉取了上一個stage的task的輸出後，進行聚合等操做時使用，默認也是佔Executor總內存的20%；第三塊是讓RDD持久化時使用，默認佔Executor總內存的60%。

　　task的執行速度是跟每一個Executor進程的CPU core數量有直接關係的。一個CPU core同一時間只能執行一個線程。而每一個Executor進程上分配到的多個task，都是以每一個task一條線程的方式，多線程併發運行的。若是CPU core數量比較充足，並且分配到的task數量比較合理，那麼一般來講，能夠比較快速和高效地執行完這些task線程。

以上就是Spark做業的基本運行原理的說明

Refer

Spark(一): 基本架構及原理

Spark 學習: spark 原理簡述與 shuffle 過程介紹