Spark架構解析(轉)

Application:

Application是建立了SparkContext實例對象的Spark用戶，包含了Driver程序, 
Spark-shell是一個應用程序，由於spark-shell在啓動的時候建立了一個SparkContext對象，其名稱爲sc,也就是說只要建立一個SparkContext就有對應的application，而一個action對應一個driver.相對應的transformation只產生元數據。

Job:

和action相對應，每個action例如count、saveAsTextFile等都會對應一個job實例，該job實例包含多任務的並行計算。

Stage：

一個job會被拆分紅不少任務，每一組任務被稱爲stage，這個MapReduce的map和reduce任務很像，劃分stage的依據在於：stage開始通常是因爲讀取外部數據或者shuffle數據、一個stage的結束通常是因爲發生shuffle（例如reduceByKey操做）或者整個job結束時，例如要把數據放到hdfs等存儲系統上

Driver Program：

運行main函數而且建立SparkContext實例的程序

Cluster Manager：

在集羣上獲取資源的外部服務 （如：standalone,Mesos,yarn）。集羣資源的管理外部服務，在spark上如今有standalone、yarn、mesos等三種集羣資源管理器，spark自帶的standalone模式可以知足大部分的spark計算環境對集羣資源管理的需求，基本上只有在集羣中運行多套計算框架的時候才考慮yarn和mesos

Worker Node：

集羣中能夠運行應用代碼的工做節點，至關於Hadoop的slave節點

Executor：

在一個Worker Node上爲應用啓動的工做進程，在進程中賦值任務的運行，而且負責將數據存放在內存或磁盤上，必須注意的是，每一個應用在一個Worker Node上只會有一個Executor，在Executor內部經過多線程的方式併發處理應用的任務。每一個應用對應獨立的executor.運行在不一樣的jvm.

Task：

被Driver送到Executor上的工做單元，一般狀況下一個task會處理一個split的數據，每一個split通常就是一個Block塊的大小。split即內存槽

stage:

一個job會被拆分紅多組任務，每組任務叫作stage(就像mapreduce分爲map任務和reduce任務)。通常stage起始於從外部讀數據，結束於shuffle階段，或任務結束。

spark架構

 
任何Spark的進程都是一個JVM進程，既然是一個JVM進程，那麼就能夠配置它的堆大小（-Xmx和-Xms）,可是進程怎麼使用堆內存和爲何須要它呢？下面是一個JVM堆空間下Spark的內存分配狀況 
 
默認狀況下，Spark進程的堆空間是512mb，爲了安全考慮同時避免OOM,Spark只容許利用90%的堆空間，spark中使用spark.storage.safetyFraction用來配置該值（默認是0.9). Spark做爲一個內存計算工具，Spark能夠在內存中存儲數據，它只是把內存做爲他的LRU緩存，這樣大量的內存被用來緩存正在計算的數據，該部分佔safe堆的60%，Spark使用spark.storage.memoryFraction控制該值，若是想知道Spark中能緩存了多少數據，能夠統計全部Executor的堆大小，乘上safeFraction和memoryFraction，默認是54%,這就是Spark可用緩存數據使用的堆大小。 
該部分介紹shuffle的內存使用狀況，它經過 堆大小 * spark.shuffle.safetyFraction * spark.shuffle.memoryFraction。 spark.shuffle.safetyFraction的默認值是0.8， spark.shuffle.memoryFraction的默認值是0.2，因此最終只能最多使堆空間的16%用於shuffle。，可是一般spark會使用這塊內存用於shuffle中一些別的任務，當執行shuffle時，有時對數據進行排序，當進行排序時，須要緩衝排完序後的數據（注意不能改變LRU緩衝中的數據，由於後面可能要重用），這樣就須要大量的RAM存儲排完序後的數據塊，當沒有足夠的內存用於排序，參考外排的實現，能夠一塊一塊的排序，而後最終合併。 
最後要講到的一塊內存是」unroll」,該快內存用於unroll計算以下：spark.storage.unrollFraction * spark.storage.memoryFraction * spark.storage.safetyFraction 。當咱們須要在內存展開數據塊的時候使用，那麼爲何須要展開呢？由於spark容許以序列化和非序列化兩種方式存儲數據，序列化後的數據沒法直接使用，因此使用時必需要展開。該部份內存佔用緩存的內存，因此若是須要內存用於展開數據時，若是這個時候內存不夠，那麼Spark LRU緩存中的數據會刪除一些快。 
此時應該清楚知道spark怎麼使用JVM中堆內存了，如今切換到集羣模式，當你啓動一個spark集羣，如何看待它，下圖是YARN模式下的 
 
當運行在yarn集羣上時，Yarn的ResourceMananger用來管理集羣資源，集羣上每一個節點上的NodeManager用來管控所在節點的資源，從yarn的角度來看，每一個節點看作可分配的資源池，當向ResourceManager請求資源時，它返回一些NodeManager信息，這些NodeManager將會提供execution container給你，每一個execution container就是知足請求的堆大小的JVM進程，JVM進程的位置是由ResourceMananger管理的，不能本身控制，若是一個節點有64GB的內存被yarn管理（經過yarn.nodemanager.resource.memory-mb配置),當請求10個4G內存的executors時，這些executors可能運行在同一個節點上。 
當在yarn上啓動spark集羣上，能夠指定executors的數量（-num-executors或者spark.executor.instances)，能夠指定每一個executor使用的內存（-executor-memory或者spark.executor.memory),能夠指定每一個executor使用的cpu核數（-executor-cores或者spark.executor.cores),指定每一個task執行使用的core數（spark.task.cpus),也能夠指定driver應用使用的內存（-driver-memory和spark.driver.memory) 
當在集羣上執行應用時，job會被切分紅stages,每一個stage切分紅task,每一個task單獨調度，能夠把executor的jvm進程看作task執行池，每一個executor有 spark.executor.cores / spark.task.cpus execution 個執行槽，這裏有個例子：集羣有12個節點運行Yarn的NodeManager，每一個節點有64G內存和32的cpu核，每一個節點能夠啓動2個executor，每一個executor的使用26G內存，剩下的內用系統和別的服務使用，每一個executor有12個cpu核用於執行task,這樣整個集羣有12 machines * 2 executors per machine * 12 cores per executor / 1 core = 288 個task執行槽，這意味着spark集羣能夠同時跑288個task,整個集羣用戶緩存數據的內存有0.9 spark.storage.safetyFraction * 0.6 spark.storage.memoryFraction * 12 machines * 2 executors per machine * 26 GB per executor = 336.96 GB. 
到目前爲止，咱們已經瞭解了spark怎麼使用JVM的內存以及集羣上執行槽是什麼，目前爲止尚未談到task的一些細節，這將在另外一個文章中提升，基本上就是spark的一個工做單元，做爲exector的jvm進程中的一個線程執行，這也是爲何spark的job啓動時間快的緣由，在jvm中啓動一個線程比啓動一個單獨的jvm進程塊（在hadoop中執行mapreduce應用會啓動多個jvm進程） 
下面將關注spark的另外一個抽象：partition, spark處理的全部數據都會切分紅partion,一個parition是什麼以及怎麼肯定，partition的大小徹底依賴數據源，spark中大部分用於讀取數據的方法均可以指定生成的RDD中partition的個數，當從hdfs上讀取一個文件時，會使用Hadoop的InputFormat來處理，默認狀況下InputFormat返回每一個InputSplit都會映射RDD中的一個Partition,大部分存儲在HDFS上的文件每一個數據塊會生成一個InputSplit,每一個數據塊大小爲64mb和128mb,由於HDFS上面的數據的塊邊界是按字節來算的（64mb一個塊），可是當數據被處理是，它又要按記錄進行切分，對於文本文件來講切分的字符就是換行符，對於sequence文件來講，他是塊結束，若是是壓縮文件，整個文件都被壓縮了，它不能按行進行切分了，整個文件只有一個inputsplit,這樣spark中也會只有一個parition,在處理的時候須要手動的repatition。

Spark架構與做業執行流程簡介

Client：客戶端進程，負責提交做業到Master。 
Master：Standalone模式中主控節點，負責接收Client提交的做業，管理Worker，並命令Worker啓動Driver和Executor。 
Worker：Standalone模式中slave節點上 的 守護進程 ，負責管理本節點的資源，按期向 Master彙報心跳，接收Master的命令，啓動Driver和Executor 
Driver： 一個Spark做業運行時包括一個Driver進程，也是做業的主進程，負責做業的解析、生成Stage並調度Task到Executor上。包括DAGScheduler，TaskScheduler。 
Executor：即真正執行做業的地方，一個集羣通常包含多個Executor，每一個Executor接收Driver的命令Launch Task，一個Executor能夠執行一到多個Task。 
Stage：一個Spark做業通常包含一到多個Stage。 
Task：一個Stage包含一到多個Task，經過多個Task實現並行運行的功能 
DAGScheduler： 實現將Spark做業分解成一到多個Stage，每一個Stage根據RDD的Partition個數決定Task的個數，而後生成相應的Task set放到TaskScheduler中。 
TaskScheduler：實現Task分配到Executor上執行。 
提交做業有兩種方式，分別是Driver（做業的master，負責做業的解析、生成stage並調度task到，包含DAGScheduler）運行在Worker上，Driver運行在客戶端。接下來分別介紹兩種方式的做業運行原理。

Driver運行在Worker上

經過org.apache.spark.deploy.Client類執行做業，做業運行命令以下：

./bin/spark-class org.apache.spark.deploy.Client launch spark://host:port file:///jar_url org.apache.spark.examples.SparkPi spark://host:port

• 1

做業執行流程描述：

一、客戶端提交做業給Master 
二、Master讓一個Worker啓動Driver，即SchedulerBackend。Worker建立一個DriverRunner線程，DriverRunner啓動SchedulerBackend進程。 
三、另外Master還會讓其他Worker啓動Exeuctor，即ExecutorBackend。Worker建立一個ExecutorRunner線程，ExecutorRunner會啓動ExecutorBackend進程。 
四、ExecutorBackend啓動後會向Driver的SchedulerBackend註冊。SchedulerBackend進程中包含DAGScheduler，它會根據用戶程序，生成執行計劃，並調度執行。對於每一個stage的task，都會被存放到TaskScheduler中，ExecutorBackend向SchedulerBackend彙報的時候把TaskScheduler中的task調度到ExecutorBackend執行。 
五、全部stage都完成後做業結束。

Driver運行在客戶端

./bin/run-example org.apache.spark.examples.SparkPi spark://host:port

• 1

 
一、客戶端啓動後直接運行用戶程序，啓動Driver相關的工做：DAGScheduler和BlockManagerMaster等。 
二、客戶端的Driver向Master註冊。 
三、Master還會讓Worker啓動Exeuctor。Worker建立一個ExecutorRunner線程，ExecutorRunner會啓動ExecutorBackend進程。 
四、ExecutorBackend啓動後會向Driver的SchedulerBackend註冊。Driver的DAGScheduler解析做業並生成相應的Stage，每一個Stage包含的Task經過TaskScheduler分配給Executor執行。 
五、全部stage都完成後做業結束。

基於Yarn的Spark架構與做業執行流程

 
這裏Spark AppMaster至關於Standalone模式下的SchedulerBackend，Executor至關於standalone的ExecutorBackend，spark AppMaster中包括DAGScheduler和YarnClusterScheduler。

refrences：

http://www.tuicool.com/articles/qaEVFb 
http://www.cnblogs.com/gaoxing/p/5041806.html 
http://blog.csdn.net/stark_summer/article/details/42833609