1.引言 

Spark 是一個基于內(nèi)存處理的計(jì)算引擎，其中任務(wù)執(zhí)行的所有計(jì)算都發(fā)生在內(nèi)存中。因此，了解 Spark 內(nèi)存管理非常重要。這將有助于我們開發(fā) Spark 應(yīng)用程序并執(zhí)行性能調(diào)優(yōu)。我們在使用spark-submit去提交spark任務(wù)的時候可以使用--executor-memory和--driver-memory這兩個參數(shù)去指定任務(wù)提交時的內(nèi)存分配，如果提交時內(nèi)存分配過大，會占用資源。如果內(nèi)存分配太小，則很容易出現(xiàn)內(nèi)存溢出和滿GC問題。

Efficient memory use is critical for good performance, but the reverse is also true: inefficient memory use leads to bad performance.

Spark 的整體架構(gòu)圖如下：

圖片

Spark 應(yīng)用程序包括兩個 JVM 進(jìn)程：driver進(jìn)程和executor進(jìn)程。其中：

• driver進(jìn)程是主控制進(jìn)程，負(fù)責(zé)創(chuàng)建 SparkSession/SparkContext、提交作業(yè)、將作業(yè)轉(zhuǎn)換為任務(wù)以及協(xié)調(diào)執(zhí)行器之間的任務(wù)執(zhí)行。
• executor進(jìn)程主要負(fù)責(zé)執(zhí)行特定的計(jì)算任務(wù)并將結(jié)果返回給驅(qū)動程序。driver的進(jìn)程的內(nèi)存管理相對簡單，Spark并沒有對此制定具體內(nèi)存管理計(jì)劃。

因此在這篇文章中，我們將會詳細(xì)深入分析executor的內(nèi)存管理。

2.Excutor內(nèi)存模型

executor充當(dāng)在工作節(jié)點(diǎn)上啟動的 JVM 進(jìn)程。因此，了解 JVM 內(nèi)存管理非常重要。我們知道JVM 內(nèi)存管理分為兩種類型： 

• 堆內(nèi)存管理（In-Heap Memory）：對象在 JVM 堆上分配并由 GC 綁定。
• 堆外內(nèi)存管理（外部內(nèi)存）：對象通過序列化在JVM外部的內(nèi)存中分配，由應(yīng)用程序管理，不受GC約束。 

整體的JVM結(jié)構(gòu)如下所示：

圖片

通常，對象的讀寫速度為：on-heap > off-heap > disk

2.1 內(nèi)存管理 

Spark 內(nèi)存管理分為兩種類型：靜態(tài)內(nèi)存管理器（Static Memory Management，SMM），以及統(tǒng)一內(nèi)存管理器（Unified Memory Management，UMM）。

圖片

在Spark1.6.0之前只有一種內(nèi)存管理方案，即Static Memory Management，但是從 Spark 1.6.0 開始，引入Unified Memory Manager 內(nèi)存管理方案，并被設(shè)置為 Spark 的默認(rèn)內(nèi)存管理器，從代碼中開始發(fā)現(xiàn)（以下代碼是基于spark 2.4.8）。

// Determine whether to use the old memory management mode
val useLegacyMemoryManager = conf.getBoolean("spark.memory.useLegacyMode", false)


val memoryManager: MemoryManager =
if (useLegacyMemoryManager) {


// The old version uses static memory management
new StaticMemoryManager(conf, numUsableCores)


} else {
// The new version uses unified memory management
UnifiedMemoryManager(conf, numUsableCores)


}

而在最新的Spark 3.x開始， Static Memory Management由于缺乏靈活性而已棄用，在源碼中已經(jīng)看到關(guān)于Static Memory Management的所有代碼，自然也就看不到控制內(nèi)存管理方案選擇的spark.memory.useLegacyMode這個參數(shù)。

2.2 靜態(tài)內(nèi)存管理器（SMM）

雖然在spark 3.x版本開始SMM已經(jīng)被淘汰了，但是目前很多企業(yè)使用的spark的版本還有很多是3.x之前的，因此我覺得為了整個學(xué)習(xí)的連貫性，還是有必要說一下的靜態(tài)內(nèi)存管理器 （SMM） 是用于內(nèi)存管理的傳統(tǒng)模型和簡單方案，該方案實(shí)現(xiàn)上簡單粗暴，將整個內(nèi)存區(qū)間分成了：存儲內(nèi)存（storage memory,）、執(zhí)行內(nèi)存（execution memory）和其他內(nèi)存（other memory）的大小在應(yīng)用程序處理過程中是固定的，但用戶可以在應(yīng)用程序啟動之前進(jìn)行配置。這三部分內(nèi)存的作用及占比如下：storage memory：主要用于緩存數(shù)據(jù)塊以提高性能，同時也用于連續(xù)不斷地廣播或發(fā)送大的任務(wù)結(jié)果。通過spark.storage.memoryFraction進(jìn)行配置，默認(rèn)為0.6。

/**
 * Return the total amount of memory available for the storage region, in bytes.
 */
private def getMaxStorageMemory(conf: SparkConf): Long = {
  val systemMaxMemory = conf.getLong("spark.testing.memory", Runtime.getRuntime.maxMemory)
  val memoryFraction = conf.getDouble("spark.storage.memoryFraction", 0.6)
  val safetyFraction = conf.getDouble("spark.storage.safetyFraction", 0.9)
  (systemMaxMemory * memoryFraction * safetyFraction).toLong
}

其中又可以分成兩部分：預(yù)留區(qū)域：這部分主要是為了防止OOM，大概占了存儲區(qū)域中的10%，由參數(shù)spark.storage.safetyFraction控制；可用的存儲區(qū)域：該區(qū)域主要是為了緩存RDD的數(shù)據(jù)和Broadcast數(shù)據(jù)，大概占了存儲區(qū)域的90%。另外該區(qū)域中并不是所有的內(nèi)存都用于以上作用，還單獨(dú)拎出來一部分區(qū)域用于緩存iterator形式的block數(shù)據(jù)，我們稱之為Unroll區(qū)域，由參數(shù)spark.storage.unrollFraction控制，大概占了可用的存儲區(qū)域的20%，如下：

private val maxUnrollMemory: Long = {
  (maxOnHeapStorageMemory * conf.getDouble("spark.storage.unrollFraction", 0.2)).toLong
}

execution memory：在執(zhí)行shuffle、join、sort和aggregation時，用于緩存中間數(shù)據(jù)。通過spark.shuffle.memoryFraction進(jìn)行配置，默認(rèn)為0.2。

private def getMaxExecutionMemory(conf: SparkConf): Long = {
  val systemMaxMemory = conf.getLong("spark.testing.memory", Runtime.getRuntime.maxMemory)

  if (systemMaxMemory < MIN_MEMORY_BYTES) {
    throw new IllegalArgumentException(s"System memory $systemMaxMemory must " +
      s"be at least $MIN_MEMORY_BYTES. Please increase heap size using the --driver-memory " +
      s"option or spark.driver.memory in Spark configuration.")
  }
  if (conf.contains("spark.executor.memory")) {
    val executorMemory = conf.getSizeAsBytes("spark.executor.memory")
    if (executorMemory < MIN_MEMORY_BYTES) {
      throw new IllegalArgumentException(s"Executor memory $executorMemory must be at least " +
        s"$MIN_MEMORY_BYTES. Please increase executor memory using the " +
        s"--executor-memory option or spark.executor.memory in Spark configuration.")
    }
  }
  val memoryFraction = conf.getDouble("spark.shuffle.memoryFraction", 0.2)
  val safetyFraction = conf.getDouble("spark.shuffle.safetyFraction", 0.8)
  (systemMaxMemory * memoryFraction * safetyFraction).toLong
}

從代碼中我們可以看到，可執(zhí)行內(nèi)存也分成了兩個部分：預(yù)留部分和可用部分，類似存儲內(nèi)存學(xué)習(xí)，這里不在贅述。other memory：除了以上兩部分的內(nèi)存，剩下的就是用于其他用作的內(nèi)存，默認(rèn)為0.2。這部分內(nèi)存用于存儲運(yùn)行Spark系統(tǒng)本身需要加載的代碼與元數(shù)據(jù)。因此，關(guān)于SMM的整體分配圖如下：

圖片

基于此就會產(chǎn)生不可逾越的缺點(diǎn)：即使存儲內(nèi)存有可用空間，我們也無法使用它，并且由于執(zhí)行程序內(nèi)存已滿，因此存在磁盤溢出。（反之亦然）。另外一個最大的問題就是：SMM只支持堆內(nèi)內(nèi)存（On-Heap），不支持對外內(nèi)存（Off-Heap）

補(bǔ)充知識1：在Spark的存儲體系中，數(shù)據(jù)的讀寫是以塊（Block）為單位，也就是說Block是Spark存儲的基本單位，這里的Block和Hdfs的Block是不一樣的，HDFS中是對大文件進(jìn)行分Block進(jìn)行存儲，Block大小是由dfs.blocksize決定的；而Spark中的Block是用戶的操作單位，一個Block對應(yīng)一塊有組織的內(nèi)存，一個完整的文件或文件的區(qū)間端，并沒有固定每個Block大小的做法。每個塊都有唯一的標(biāo)識，Spark把這個標(biāo)識抽象為BlockId。BlockId本質(zhì)上是一個字符串，但是在Spark中將它保證為"一組"case類，這些類的不同本質(zhì)是BlockID這個命名字符串的不同，從而可以通過BlockID這個字符串來區(qū)別BlockId

補(bǔ)充知識2：內(nèi)存池是Spark內(nèi)存的抽象，它記錄了總內(nèi)存大小，已使用內(nèi)存大小，剩余內(nèi)存大小，提供給MemoryManager進(jìn)行分配/回收內(nèi)存。它包括兩個實(shí)現(xiàn)類：ExecutionMemoryPool和StorageMemoryPool，分別對應(yīng)execution memory和storage memory。當(dāng)需要新的內(nèi)存時，spark通過memoryPool來判斷內(nèi)存是否充足。需要注意的是memoryPool以及子類方法只是用來標(biāo)記內(nèi)存使用情況，而不實(shí)際分配/回收內(nèi)存。 

2.3 統(tǒng)一內(nèi)存管理器（UMM）

從 Spark 1.6.0 開始，采用了新的內(nèi)存管理器來取代靜態(tài)內(nèi)存管理器，并為 Spark 提供動態(tài)內(nèi)存分配。它將內(nèi)存區(qū)域分配為由存儲和執(zhí)行共享的統(tǒng)一內(nèi)存容器。當(dāng)未使用執(zhí)行內(nèi)存時，存儲內(nèi)存可以獲取所有可用內(nèi)存，反之亦然。如果任何存儲或執(zhí)行內(nèi)存需要更多空間，則會調(diào)用acquireMemory方法將擴(kuò)展其中一個內(nèi)存池并收縮另一個內(nèi)存池。因此，UMM相比SMM的內(nèi)存管理優(yōu)勢明顯：存儲內(nèi)存和執(zhí)行內(nèi)存之間的邊界不是靜態(tài)的，在內(nèi)存壓力的情況下，邊界會移動，即一個區(qū)域會通過從另一個區(qū)域借用空間來增長。當(dāng)應(yīng)用程序沒有緩存并且正在進(jìn)行時，執(zhí)行會使用所有內(nèi)存以避免不必要的磁盤溢出。當(dāng)應(yīng)用程序有緩存時，它將保留最小存儲內(nèi)存，以便數(shù)據(jù)塊不受影響。此內(nèi)存管理可為各種工作負(fù)載提供合理的開箱即用性能，而無需用戶了解內(nèi)存內(nèi)部劃分方式的專業(yè)知識。

2.3.1 堆內(nèi)存

默認(rèn)情況下，Spark 僅使用堆內(nèi)存。Spark 應(yīng)用程序啟動時，堆內(nèi)存的大小由 --executor-memory 或 spark.executor.memory 參數(shù)配置。在UMM下，spark的堆內(nèi)存結(jié)構(gòu)圖如下：

圖片

我們發(fā)現(xiàn)大體上和SMM沒有太大的區(qū)別，包括每個區(qū)域的功能，只是UMM在Storage和Execution可以彈性的變化（這一點(diǎn)也是spark rdd中“彈性”的體現(xiàn)之一）。

備注：在 Spark 1.6 中，spark.memory.fraction 值為 0.75，spark.memory.storageFraction 值為 0.5。從spark 2.x開始spark.memory.fraction 值為 0.6。

2.3.1.1 System Reserved：系統(tǒng)預(yù)留

預(yù)留內(nèi)存是為系統(tǒng)預(yù)留的內(nèi)存，用于存儲Spark的內(nèi)部對象。從 Spark 1.6 開始，該值為 300MB。這意味著 300MB 的 RAM 不參與 Spark 內(nèi)存區(qū)域大小計(jì)算。預(yù)留內(nèi)存的大小是硬編碼的，如果不重新編譯 Spark 或設(shè)置 spark.testing.reservedMemory，則無法以任何方式更改其大小，一般在實(shí)際的生產(chǎn)環(huán)境中不建議修改此值。

private val RESERVED_SYSTEM_MEMORY_BYTES = 300 * 1024 * 1024


val reservedMemory = conf.getLong("spark.testing.reservedMemory",
  if (conf.contains("spark.testing")) 0 else RESERVED_SYSTEM_MEMORY_BYTES)
val minSystemMemory = (reservedMemory * 1.5).ceil.toLong
if (systemMemory < minSystemMemory) {
  throw new IllegalArgumentException(s"System memory $systemMemory must " +
    s"be at least $minSystemMemory. Please increase heap size using the --driver-memory " +
    s"option or spark.driver.memory in Spark configuration.")
}

從源碼中我們可以看出，如果執(zhí)行程序內(nèi)存小于保留內(nèi)存的 1.5 倍（1.5 * 保留內(nèi)存 = 450MB），則 Spark 作業(yè)將失敗，并顯示以下異常消息：

24/03/20 13:55:51 ERROR repl.Main: Failed to initialize Spark session.
java.lang.IllegalArgumentException: Executor memory 314572800 must be at least 471859200. Please increase executor memory using the --executor-memory option or spark.executor.memory in Spark configuration.
        at org.apache.spark.memory.UnifiedMemoryManager$.getMaxMemory(UnifiedMemoryManager.scala:225)
        at org.apache.spark.memory.UnifiedMemoryManager$.apply(UnifiedMemoryManager.scala:199)

2.3.1.2 其他內(nèi)存（或稱用戶內(nèi)存）

其他內(nèi)存是用于存儲用戶定義的數(shù)據(jù)結(jié)構(gòu)、Spark 內(nèi)部元數(shù)據(jù)、用戶創(chuàng)建的任何 UDF 以及 RDD 轉(zhuǎn)換操作所需的數(shù)據(jù)（如 RDD 依賴信息等）的內(nèi)存。例如，我們可以通過使用 mapPartitions 轉(zhuǎn)換來重寫 Spark 聚合，以維護(hù)一個哈希表以運(yùn)行此聚合，這將消耗所謂的其他內(nèi)存。此內(nèi)存段不受 Spark 管理，計(jì)算公式為：(Java Heap - Reserved Memory) * (1.0 - spark.memory.fraction)。

2.3.1.3 Spark內(nèi)存（或稱統(tǒng)一內(nèi)存）

Spark Memory 是由 Apache Spark 管理的內(nèi)存池。Spark Memory 負(fù)責(zé)在執(zhí)行任務(wù)（如聯(lián)接）或存儲廣播變量時存儲中間狀態(tài)。計(jì)算公式為：(Java Heap - Reserved Memory) * spark.memory.fraction。

Spark 任務(wù)在兩個主要內(nèi)存區(qū)域中運(yùn)行：

• Executor Memory：用于隨機(jī)播放、聯(lián)接、排序和聚合。
• Storage Memory：用于緩存數(shù)據(jù)分區(qū)。

它們之間的邊界由 spark.memory.storageFraction 參數(shù)設(shè)置，默認(rèn)為 0.5 或 50%。

1）StorageMemory: 存儲內(nèi)存

存儲內(nèi)存用于存儲所有緩存數(shù)據(jù)、廣播變量、unroll數(shù)據(jù)等，“unroll”本質(zhì)上是反序列化序列化數(shù)據(jù)的過程。任何包含內(nèi)存的持久性選項(xiàng)都會將該數(shù)據(jù)存儲在此段中。Spark 通過刪除基于最近最少使用 （LRU） 機(jī)制的舊緩存對象來為新緩存請求清除空間。緩存的數(shù)據(jù)從存儲中取出后，將寫入磁盤或根據(jù)配置重新計(jì)算。廣播變量存儲在緩存中，具有MEMORY_AND_DISK持久性級別。這就是我們存儲緩存數(shù)據(jù)的地方，這些數(shù)據(jù)是長期存在的。

計(jì)算公式：

(Java Heap - Reserved Memory) * spark.memory.fraction * spark.memory.storageFraction

2）Execution Memory：執(zhí)行內(nèi)存

執(zhí)行內(nèi)存用于存儲 Spark 任務(wù)執(zhí)行過程中所需的對象。例如，它用于將映射端的shuffle中間緩沖區(qū)存儲在內(nèi)存中。此外，它還用于存儲hash聚合步驟的hash table。如果沒有足夠的可用內(nèi)存，執(zhí)行內(nèi)存池還支持溢出磁盤，但是其他線程（任務(wù)）無法強(qiáng)制逐出此池中的block。執(zhí)行內(nèi)存往往比存儲內(nèi)存壽命更短。每次操作后都會立即將其逐出，為下一次操作騰出空間。

計(jì)算公式：

(Java Heap - Reserved Memory) * spark.memory.fraction * (1.0 -  spark.memory.storageFraction)

由于執(zhí)行內(nèi)存的性質(zhì)，無法從此池中強(qiáng)制逐出塊;否則，執(zhí)行將中斷，因?yàn)檎也坏剿玫膲K。但是，當(dāng)涉及到存儲內(nèi)存時，可以根據(jù)需要從內(nèi)存中逐出block并寫入磁盤或重新計(jì)算（如果持久性級別為MEMORY_ONLY）。

存儲和執(zhí)行池借用規(guī)則：

• 只有當(dāng)執(zhí)行內(nèi)存中有未使用的塊時，存儲內(nèi)存才能從執(zhí)行內(nèi)存中借用空間。
• 如果塊未在存儲內(nèi)存中使用，則執(zhí)行內(nèi)存也可以從存儲內(nèi)存中借用空間。
• 如果存儲內(nèi)存使用執(zhí)行內(nèi)存中的塊，并且執(zhí)行需要更多內(nèi)存，則可以強(qiáng)制逐出存儲內(nèi)存占用的多余塊
• 如果存儲內(nèi)存中的塊被執(zhí)行內(nèi)存使用，而存儲需要更多的內(nèi)存，則無法強(qiáng)行逐出執(zhí)行內(nèi)存占用的多余塊；它將具有更少的內(nèi)存區(qū)域。它將等到 Spark 釋放存儲在執(zhí)行內(nèi)存中的多余塊，然后占用它們。

案例：計(jì)算 5 GB 執(zhí)行程序內(nèi)存的內(nèi)存

為了計(jì)算預(yù)留內(nèi)存、用戶內(nèi)存、spark內(nèi)存、存儲內(nèi)存和執(zhí)行內(nèi)存，我們將使用以下參數(shù)：

spark.executor.memory=5g
spark.memory.fractinotallow=0.6
spark.memory.storageFractinotallow=0.5

那么會得到如下結(jié)論：

Java Heap Memory = 5 GB
                               = 5 * 1024 MB
                               = 5120 MB


Reserved Memory      = 300 MB


Usable Memory        = (Java Heap Memory - Reserved Memory)
                                 = 5120 MB - 300 MB
                                 = 4820 MB


User Memory            = Usable Memory * (1.0 * spark.memory.fraction) 
                                 = 4820 MB * (1.0 - 0.6) 
                                 = 4820 MB * 0.4 
                                 = 1928 MB


Spark Memory           = Usable Memory * spark.memory.fraction
                                  = 4820 MB * 0.6 
                                  = 2892 MB


Spark Storage Memory   = Spark Memory * Spark.memory.storageFraction
                                       = 2892 MB * 0.5 
                                       = 1446 MB


Spark Execution Memory = Spark Memory * (1.0 - spark.memory.storageFraction)
                       = 2892 MB * ( 1 - 0.5) 
                       = 2892 MB * 0.5 
                       = 1446 MB

2.3.2 堆外內(nèi)存

堆外內(nèi)存是指將內(nèi)存對象（序列化為字節(jié)數(shù)組）分配給 JVM堆之外的內(nèi)存，該堆由操作系統(tǒng)（而不是JVM）直接管理，但存儲在進(jìn)程堆之外的本機(jī)內(nèi)存中（因此，它們不會被垃圾回收器處理）。這樣做的結(jié)果是保留較小的堆，以減少垃圾回收對應(yīng)用程序的影響。訪問此數(shù)據(jù)比訪問堆存儲稍慢，但仍比從磁盤讀取/寫入快。缺點(diǎn)是用戶必須手動處理管理分配的內(nèi)存。此模型不適用于 JVM 內(nèi)存，而是將 malloc()  中不安全相關(guān)語言（如 C）的 Java API 直接調(diào)用操作系統(tǒng)以獲取內(nèi)存。由于此方法不是對 JVM 內(nèi)存進(jìn)行管理，因此請避免頻繁 GC。此應(yīng)用程序的缺點(diǎn)是內(nèi)存必須寫入自己的邏輯和內(nèi)存應(yīng)用程序版本。Spark 1.6+ 開始引入堆外內(nèi)存，可以選擇使用堆外內(nèi)存來分配 Unified Memory Manager。

默認(rèn)情況下，堆外內(nèi)存是禁用的，但我們可以通過 spark.memory.offHeap.enabled（默認(rèn)為 false）參數(shù)啟用它，并通過 spark.memory.offHeap.size（默認(rèn)為 0）參數(shù)設(shè)置內(nèi)存大小。如：

spark-shell \
    --conf spark.memory.offHeap.enabled=true \
    --conf spark.memory.offHeap.size=5g

堆外內(nèi)存支持OFF_HEAP持久性級別。與堆上內(nèi)存相比，堆外內(nèi)存的模型相對簡單，僅包括存儲內(nèi)存和執(zhí)行內(nèi)存。

如果啟用了堆外內(nèi)存，Executor 中的 Execution Memory 是堆內(nèi)的 Execution 內(nèi)存和堆外的 Execution 內(nèi)存之和。存儲內(nèi)存也是如此。

總之，Spark內(nèi)存管理的核心目標(biāo)是在有限的內(nèi)存資源下，實(shí)現(xiàn)數(shù)據(jù)緩存的最大化利用和執(zhí)行計(jì)算的高效進(jìn)行，同時盡量減少由于內(nèi)存不足導(dǎo)致的數(shù)據(jù)重算或內(nèi)存溢出等問題，是整個spark允許可以穩(wěn)定運(yùn)行的基礎(chǔ)保障。