Spark 架構

本文轉之Pivotal的一個工程師的博客。以爲極好。

 

做者本人常常在StackOverflow上回答一個關係Spark架構的問題，發現整個互聯網都沒有一篇文章能對Spark整體架構進行很好的描述，做者可憐咱們這些菜鳥，寫了這篇文章，太感動了。 本文讀者須要必定的Spark的基礎知識，至少了解Spark的RDD和DAG。

上圖引入了不少術語："Executor","Task","Cache","Worker Node"等等，當我開始學習Spark的時候，這幾乎是整個互聯網上惟一一張關於Spark架構的圖了，我我的以爲該圖缺失了一些很重要的概念或者是描述的

 

任何Spark的進程都是一個JVM進程，既然是一個JVM進程，那麼就能夠配置它的堆大小（-Xmx和-Xms）,可是進程怎麼使用堆內存和爲何須要它呢？下面是一個JVM堆空間下Spark的內存分配狀況

默認狀況下，Spark進程的堆空間是512mb，爲了安全考慮同時避免OOM,Spark只容許利用90%的堆空間，spark中使用spark.storage.safetyFraction用來配置該值（默認是0.9). Spark做爲一個內存計算工具，Spark能夠在內存中存儲數據，若是讀過 http://0x0fff.com/spark-misconceptions/ ， 就會明白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,關於怎麼使用這塊內存，參考https://github.com/apache/spark/blob/branch-1.3/core/src/main/scala/org/apache/spark/shuffle/ShuffleMemoryManager.scala ，可是一般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。

 

 

 

 

 

來自爲知筆記(Wiz)