Spark體系架構必讀
最近看到一篇關于Spark架構的博文，作者是 Alexey Grishchenko。看過Alexey博文的同學應該都知道，他對Spark理解地非常深入，讀完他的 “spark-architecture” 這篇博文，有種醍醐灌頂的感覺，從JVM內存分配到Spark集群的資源管理，步步深入，感觸頗多。因此，在周末的業余時間里，將此文的核心內容譯成中文，并在這里與大家分享。如在翻譯過程中有文字上的表達紕漏，還請大家指出。
首先來看一張Spark 1.3.0 官方給出的圖片，如下：
在這張圖中，你會看到很多的術語 ，諸如“executor”, “task”, “cache”, “Worker Node” 等。原作者表示，在他開始學spark的時候，上述圖是唯一一張可以找到的圖片(Spark 1.3.0)，形勢很不樂觀。更加不幸地是，這張圖并沒有很好地表達出Spark內在的一些概念。因此，通過不斷地學習，作者將自己所學的知識整理成一個系列，而此文僅是其中的一篇。下面進入核心要點。
Spark 內存分配
在你的cluster或是local machine上正常運行的任何Spark程序都是一個JVM進程。對于任何的JVM進程，你都可以使用-Xmx和-Xms配置它的堆大小(heap size)。問題是：這些進程是如何使用它的堆內存(heap memory)以及為何需要它呢?下面圍繞這個問題慢慢展開。
首先來看看下面這張Spark JVM堆內存分配圖：
Heap Size
默認情況下，Spark啟動時會初始化512M的JVM 堆內存。處于安全角度以及避免OOM錯誤，Spark只允許使用90%的的堆內存，該參數可以通過Spark的spark.storage.safetyFraction參數進行控制。 OK，你可能聽說Spark是基于內存的工具，它允許你將數據存在內存中。如果你讀過作者的 Spark Misconceptions 這篇文章，那么你應該知道Spark其實不是真正的基于內存(in-memory)的工具。它僅僅是在LRU cache (http://en.wikipedia.org/wiki/Cache_algorithms) 過程中使用內存。所以一部分的內存用在數據緩存上，這部分通常占安全堆內存(90%)的60%，該參數也可以通過配置spark.storage.memoryFraction進行控制。因此，如果你想知道在Spark中可以緩存多少數據，你可以通過對所有executor的堆大小求和，然后乘以safetyFraction 和storage.memoryFraction即可，默認情況下是0.9 * 0.6 = 0.54,即總的堆內存的54%可供Spark使用。
Shuffle Memory
接下來談談shuffle memory，計算公式是 “Heap Size” * spark.shuffle.safetyFraction * spark.shuffle.memoryFraction。spark.shuffle.safetyFraction的默認值是 0.8 或80%， spark.shuffle.memoryFraction的默認值是0.2或20%，所以你***可以用于shuffle的JVM heap 內存大小是 0.8*0.2=0.16，即總heap size的16%。 問題是Spark是如何來使用這部分內存呢?官方的Github上面有更詳細的解釋(https://github.com/apache/spark/blob/branch-1.3/core/src/main/scala/org/apache/spark/shuffle/ShuffleMemoryManager.scala)。
總得來說，Spark將這部分memory 用于Shuffle階段調用其他的具體task。當shuffle執行之后，有時你需要對數據進行sort。在sort階段，通常你還需要一個類似緩沖的buffer來存儲已經排序好的數據(謹記，不能修改已經LRU cache中的數據，因為這些數據可能會再次使用)。因此，需要一定數量的RAM來存儲已經sorted的數據塊。如果你沒有足夠的memory用來排序，該怎么做呢?在wikipedia 搜一下“external sorting” (外排序)，仔細研讀一下即可。外排序允許你對塊對數據塊進行分類，然后將***的結果合并到一起。
unroll Memory
關于RAM***要講到”unroll” memory，用于unroll 進程的內存總量計算公式為：spark.storage.unrollFraction * spark.storage.memoryFraction *spark.storage.safetyFraction。默認情況下是 0.2 * 0.6 * 0.9 = 0.108,即10.8%的heap size。 當你需要在內存中將數據塊展開的時候使用它。為什么需要 unroll 操作呢?在Spark中，允許以 序列化(serialized )和反序列化(deserialized) 兩種方式存儲數據，而對于序列化后的數據是無法直接使用的，所以在使用時必須對其進行unroll操作，因此這部分RAM是用于unrolling操作的內存。unroll memory 與storage RAM 是共享的，也就是當你在對數據執行unroll操作時，如果需要內存，而這個時候內存卻不夠，那么可能會致使撤銷存儲在 Spark LRU cache中少些數據塊。
Spark 集群模式JVM分配
OK，通過上面的講解，我們應該對Spark進程有了進一步的理解，并且已經知道它是如何利用JVM進程中的內存。現在切換到集群上，以YARN模式為例。
在YARN集群里，它有一個YARN ResourceMananger 守護進程控制著集群資源(也就是memory)，還有一系列運行在集群各個節點的YARN Node Managers控制著節點資源的使用。從YARN的角度來看，每個節點可以看做是可分配的RAM池，當你向ResourceManager發送request請求資源時，它會返回一些NodeManager信息，這些NodeManager將會為你提供execution container，而每個execution container 都是一個你發送請求時指定的heap size的JVM進程。JVM的位置是由 YARN ResourceMananger 管理的，你沒有控制權限。如果某個節點有64GB的RAM被YARN控制著(可通過設置yarn-site.xml 配置文件中參數 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 會給你的task設置 spark.executor.cores/ spark.task.cpus execution個執行槽。例如，在集群的YARN NodeManager中運行有12個節點，每個節點有64G內存和32個CPU核(16個超線程物理core)。每個節點可以啟動2個26G內存的executor(剩下的RAM用于系統程序、YARN NM 和DataNode)，每個executor有12個cpu核可以用于執行task(剩下的用于系統程序、YARN NM 和DataNode)，這樣整個集群可以處理 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 進程中的一個thread執行。這也是為什么Spark job啟動時間快的原因，在JVM中啟動一個線程比啟動一個單獨的JVM進程塊，而在Hadoop中執行MapReduce應用會啟動多個JVM進程。
Spark Partition
下面來談談Spark的另一個抽象概念”partition”。在Spark程序運行過程中，所有的數據都會被切分成多個Partion。問題是一個parition是什么并且如何決定partition的數量呢?首先Partition的大小完全依賴于你的數據源。在Spark中，大部分用于讀取數據的method都可以指定生成的RDD中Partition數量。當你從hdfs上讀取一個文件時，你會使用Hadoop的InputFormat來指定，默認情況下InputFormat返回每個InputSplit都會映射到RDD中的一個Partition上。對于HDFS上的大部分文件，每個數據塊都會生成一個InputSplit，大小近似為64 MB/128 MB的數據。近似情況下，HDFS上數據的塊邊界是按字節來算的(64MB一個塊)，但是當數據被處理時，它會按記錄進行切分。對于文本文件來說切分的字符就是換行符，對于sequence文件，它以塊結束等等。比較特殊的是壓縮文件，由于整個文件被壓縮了，因此不能按行進行切分了，整個文件只有一個inputsplit，這樣spark中也會只有一個parition，在處理的時候需要手動對它進行repatition。
