最近看到一篇關(guān)于Spark架構(gòu)的博文，作者是 Alexey Grishchenko?？催^Alexey博文的同學(xué)應(yīng)該都知道，他對Spark理解地非常深入，讀完他的 “spark-architecture” 這篇博文，有種醍醐灌頂?shù)母杏X，從JVM內(nèi)存分配到Spark集群的資源管理，步步深入，感觸頗多。因此，在周末的業(yè)余時間里，將此文的核心內(nèi)容譯成中文，并在這里與大家分享。如在翻譯過程中有文字上的表達紕漏，還請大家指出。

首先來看一張Spark 1.3.0 官方給出的圖片，如下：

在這張圖中，你會看到很多的術(shù)語 ，諸如“executor”, “task”, “cache”, “Worker Node” 等。原作者表示，在他開始學(xué)spark的時候，上述圖是唯一一張可以找到的圖片(Spark 1.3.0)，形勢很不樂觀。更加不幸地是，這張圖并沒有很好地表達出Spark內(nèi)在的一些概念。因此，通過不斷地學(xué)習(xí)，作者將自己所學(xué)的知識整理成一個系列，而此文僅是其中的一篇。下面進入核心要點。

Spark 內(nèi)存分配

在你的cluster或是local machine上正常運行的任何Spark程序都是一個JVM進程。對于任何的JVM進程，你都可以使用-Xmx和-Xms配置它的堆大小(heap size)。問題是：這些進程是如何使用它的堆內(nèi)存(heap memory)以及為何需要它呢?下面圍繞這個問題慢慢展開。

首先來看看下面這張Spark JVM堆內(nèi)存分配圖：

Heap Size

默認情況下，Spark啟動時會初始化512M的JVM 堆內(nèi)存。處于安全角度以及避免OOM錯誤，Spark只允許使用90%的的堆內(nèi)存，該參數(shù)可以通過Spark的spark.storage.safetyFraction參數(shù)進行控制。 OK，你可能聽說Spark是基于內(nèi)存的工具，它允許你將數(shù)據(jù)存在內(nèi)存中。如果你讀過作者的 Spark Misconceptions 這篇文章，那么你應(yīng)該知道Spark其實不是真正的基于內(nèi)存(in-memory)的工具。它僅僅是在LRU cache (http://en.wikipedia.org/wiki/Cache_algorithms) 過程中使用內(nèi)存。所以一部分的內(nèi)存用在數(shù)據(jù)緩存上，這部分通常占安全堆內(nèi)存(90%)的60%，該參數(shù)也可以通過配置spark.storage.memoryFraction進行控制。因此，如果你想知道在Spark中可以緩存多少數(shù)據(jù)，你可以通過對所有executor的堆大小求和，然后乘以safetyFraction 和storage.memoryFraction即可，默認情況下是0.9 * 0.6 = 0.54,即總的堆內(nèi)存的54%可供Spark使用。

Shuffle Memory

接下來談?wù)剆huffle memory，計算公式是 “Heap Size” * spark.shuffle.safetyFraction * spark.shuffle.memoryFraction。spark.shuffle.safetyFraction的默認值是 0.8 或80%， spark.shuffle.memoryFraction的默認值是0.2或20%，所以你***可以用于shuffle的JVM heap 內(nèi)存大小是 0.8*0.2=0.16，即總heap size的16%。 問題是Spark是如何來使用這部分內(nèi)存呢?官方的Github上面有更詳細的解釋(https://github.com/apache/spark/blob/branch-1.3/core/src/main/scala/org/apache/spark/shuffle/ShuffleMemoryManager.scala)。

總得來說，Spark將這部分memory 用于Shuffle階段調(diào)用其他的具體task。當(dāng)shuffle執(zhí)行之后，有時你需要對數(shù)據(jù)進行sort。在sort階段，通常你還需要一個類似緩沖的buffer來存儲已經(jīng)排序好的數(shù)據(jù)(謹記，不能修改已經(jīng)LRU cache中的數(shù)據(jù)，因為這些數(shù)據(jù)可能會再次使用)。因此，需要一定數(shù)量的RAM來存儲已經(jīng)sorted的數(shù)據(jù)塊。如果你沒有足夠的memory用來排序，該怎么做呢?在wikipedia 搜一下“external sorting” (外排序)，仔細研讀一下即可。外排序允許你對塊對數(shù)據(jù)塊進行分類，然后將***的結(jié)果合并到一起。

unroll Memory

關(guān)于RAM***要講到”unroll” memory，用于unroll 進程的內(nèi)存總量計算公式為：spark.storage.unrollFraction * spark.storage.memoryFraction *spark.storage.safetyFraction。默認情況下是 0.2 * 0.6 * 0.9 = 0.108,即10.8%的heap size。 當(dāng)你需要在內(nèi)存中將數(shù)據(jù)塊展開的時候使用它。為什么需要 unroll 操作呢?在Spark中，允許以 序列化(serialized )和反序列化(deserialized) 兩種方式存儲數(shù)據(jù)，而對于序列化后的數(shù)據(jù)是無法直接使用的，所以在使用時必須對其進行unroll操作，因此這部分RAM是用于unrolling操作的內(nèi)存。unroll memory 與storage RAM 是共享的，也就是當(dāng)你在對數(shù)據(jù)執(zhí)行unroll操作時，如果需要內(nèi)存，而這個時候內(nèi)存卻不夠，那么可能會致使撤銷存儲在 Spark LRU cache中少些數(shù)據(jù)塊。

Spark 集群模式JVM分配

OK，通過上面的講解，我們應(yīng)該對Spark進程有了進一步的理解，并且已經(jīng)知道它是如何利用JVM進程中的內(nèi)存?，F(xiàn)在切換到集群上，以YARN模式為例。

在YARN集群里，它有一個YARN ResourceMananger 守護進程控制著集群資源(也就是memory)，還有一系列運行在集群各個節(jié)點的YARN Node Managers控制著節(jié)點資源的使用。從YARN的角度來看，每個節(jié)點可以看做是可分配的RAM池，當(dāng)你向ResourceManager發(fā)送request請求資源時，它會返回一些NodeManager信息，這些NodeManager將會為你提供execution container，而每個execution container 都是一個你發(fā)送請求時指定的heap size的JVM進程。JVM的位置是由 YARN ResourceMananger 管理的，你沒有控制權(quán)限。如果某個節(jié)點有64GB的RAM被YARN控制著(可通過設(shè)置yarn-site.xml 配置文件中參數(shù) yarn.nodemanager.resource.memory-mb )，當(dāng)你請求10個4G內(nèi)存的executors時，這些executors可能運行在同一個節(jié)點上，即便你的集群跟大也無濟于事。

當(dāng)以YARN模式啟動spark集群時，你可以指定executors的數(shù)量(-num-executors 或者 spark.executor.instances 參數(shù))，可以指定每個executor 固有的內(nèi)存大小(-executor-memory 或者 spark.executor.memory)，可以指定每個executor使用的cpu核數(shù)(-executor-cores 或者 spark.executor.cores)，可以指定分配給每個task的core的數(shù)量(spark.task.cpus)，還可以指定 driver 上使用的內(nèi)存(-driver-memory 或者spark.driver.memory)。

當(dāng)你在集群上執(zhí)行應(yīng)用程序時，job程序會被切分成多個stages，每個stage又會被切分成多個task，每個task單獨調(diào)度，可以把每個executor的JVM進程看做是task執(zhí)行槽池，每個executor 會給你的task設(shè)置 spark.executor.cores/ spark.task.cpus execution個執(zhí)行槽。例如，在集群的YARN NodeManager中運行有12個節(jié)點，每個節(jié)點有64G內(nèi)存和32個CPU核(16個超線程物理core)。每個節(jié)點可以啟動2個26G內(nèi)存的executor(剩下的RAM用于系統(tǒng)程序、YARN NM 和DataNode)，每個executor有12個cpu核可以用于執(zhí)行task(剩下的用于系統(tǒng)程序、YARN NM 和DataNode)，這樣整個集群可以處理 12 machines * 2 executors per machine * 12 cores per executor / 1 core = 288 個task 執(zhí)行槽，這意味著你的spark集群可以同時跑288個task，幾乎充分利用了所有的資源。整個集群用于緩存數(shù)據(jù)的內(nèi)存有0.9 spark.storage.safetyFraction * 0.6 spark.storage.memoryFraction * 12 machines * 2 executors per machine * 26 GB per executor = 336.96 GB. 實際上沒有那么多，但在大多數(shù)情況下，已經(jīng)足夠了。

到這里，大概已經(jīng)了解了spark是如何使用JVM的內(nèi)存，并且知道什么是集群的執(zhí)行槽。而關(guān)于task，它是Spark執(zhí)行的工作單元，并且作為exector JVM 進程中的一個thread執(zhí)行。這也是為什么Spark job啟動時間快的原因，在JVM中啟動一個線程比啟動一個單獨的JVM進程塊，而在Hadoop中執(zhí)行MapReduce應(yīng)用會啟動多個JVM進程。

Spark Partition

下面來談?wù)凷park的另一個抽象概念”partition”。在Spark程序運行過程中，所有的數(shù)據(jù)都會被切分成多個Partion。問題是一個parition是什么并且如何決定partition的數(shù)量呢?首先Partition的大小完全依賴于你的數(shù)據(jù)源。在Spark中，大部分用于讀取數(shù)據(jù)的method都可以指定生成的RDD中Partition數(shù)量。當(dāng)你從hdfs上讀取一個文件時，你會使用Hadoop的InputFormat來指定，默認情況下InputFormat返回每個InputSplit都會映射到RDD中的一個Partition上。對于HDFS上的大部分文件，每個數(shù)據(jù)塊都會生成一個InputSplit，大小近似為64 MB/128 MB的數(shù)據(jù)。近似情況下，HDFS上數(shù)據(jù)的塊邊界是按字節(jié)來算的(64MB一個塊)，但是當(dāng)數(shù)據(jù)被處理時，它會按記錄進行切分。對于文本文件來說切分的字符就是換行符，對于sequence文件，它以塊結(jié)束等等。比較特殊的是壓縮文件，由于整個文件被壓縮了，因此不能按行進行切分了，整個文件只有一個inputsplit，這樣spark中也會只有一個parition，在處理的時候需要手動對它進行repatition。