Spark 作為一個(gè)基于內(nèi)存的分布式計(jì)算引擎，其內(nèi)存管理模塊在整個(gè)系統(tǒng)中扮演著非常重要的角色。理解 Spark 內(nèi)存管理的基本原理，有助于更好地開發(fā) Spark 應(yīng)用程序和進(jìn)行性能調(diào)優(yōu)。本文旨在梳理出 Spark 內(nèi)存管理的脈絡(luò)，拋磚引玉，引出讀者對(duì)這個(gè)話題的深入探討。本文中闡述的原理基于 Spark 2.1 版本，閱讀本文需要讀者有一定的 Spark 和 Java 基礎(chǔ)，了解 RDD、Shuffle、JVM 等相關(guān)概念。

在執(zhí)行 Spark 的應(yīng)用程序時(shí)，Spark 集群會(huì)啟動(dòng) Driver 和 Executor 兩種 JVM 進(jìn)程，前者為主控進(jìn)程，負(fù)責(zé)創(chuàng)建 Spark 上下文，提交 Spark 作業(yè)（Job），并將作業(yè)轉(zhuǎn)化為計(jì)算任務(wù)（Task），在各個(gè) Executor 進(jìn)程間協(xié)調(diào)任務(wù)的調(diào)度，后者負(fù)責(zé)在工作節(jié)點(diǎn)上執(zhí)行具體的計(jì)算任務(wù)，并將結(jié)果返回給 Driver，同時(shí)為需要持久化的 RDD 提供存儲(chǔ)功能。由于 Driver 的內(nèi)存管理相對(duì)來說較為簡(jiǎn)單，本文主要對(duì) Executor 的內(nèi)存管理進(jìn)行分析，下文中的 Spark 內(nèi)存均特指 Executor 的內(nèi)存。

1. 堆內(nèi)和堆外內(nèi)存規(guī)劃

作為一個(gè) JVM 進(jìn)程，Executor 的內(nèi)存管理建立在 JVM 的內(nèi)存管理之上，Spark 對(duì) JVM 的堆內(nèi)（On-heap）空間進(jìn)行了更為詳細(xì)的分配，以充分利用內(nèi)存。同時(shí)，Spark 引入了堆外（Off-heap）內(nèi)存，使之可以直接在工作節(jié)點(diǎn)的系統(tǒng)內(nèi)存中開辟空間，進(jìn)一步優(yōu)化了內(nèi)存的使用。

圖 1 . 堆內(nèi)和堆外內(nèi)存示意圖

1.1 堆內(nèi)內(nèi)存

堆內(nèi)內(nèi)存的大小，由 Spark 應(yīng)用程序啟動(dòng)時(shí)的 –executor-memory 或 spark.executor.memory 參數(shù)配置。Executor 內(nèi)運(yùn)行的并發(fā)任務(wù)共享 JVM 堆內(nèi)內(nèi)存，這些任務(wù)在緩存 RDD 數(shù)據(jù)和廣播（Broadcast）數(shù)據(jù)時(shí)占用的內(nèi)存被規(guī)劃為存儲(chǔ)（Storage）內(nèi)存，而這些任務(wù)在執(zhí)行 Shuffle 時(shí)占用的內(nèi)存被規(guī)劃為執(zhí)行（Execution）內(nèi)存，剩余的部分不做特殊規(guī)劃，那些 Spark 內(nèi)部的對(duì)象實(shí)例，或者用戶定義的 Spark 應(yīng)用程序中的對(duì)象實(shí)例，均占用剩余的空間。不同的管理模式下，這三部分占用的空間大小各不相同（下面第 2 小節(jié)會(huì)進(jìn)行介紹）。

Spark 對(duì)堆內(nèi)內(nèi)存的管理是一種邏輯上的”規(guī)劃式”的管理，因?yàn)閷?duì)象實(shí)例占用內(nèi)存的申請(qǐng)和釋放都由 JVM 完成，Spark 只能在申請(qǐng)后和釋放前記錄這些內(nèi)存，我們來看其具體流程：

• 申請(qǐng)內(nèi)存：

1. Spark 在代碼中 new 一個(gè)對(duì)象實(shí)例
2. JVM 從堆內(nèi)內(nèi)存分配空間，創(chuàng)建對(duì)象并返回對(duì)象引用
3. Spark 保存該對(duì)象的引用，記錄該對(duì)象占用的內(nèi)存

• 釋放內(nèi)存：

1. Spark 記錄該對(duì)象釋放的內(nèi)存，刪除該對(duì)象的引用
2. 等待 JVM 的垃圾回收機(jī)制釋放該對(duì)象占用的堆內(nèi)內(nèi)存

我們知道，JVM 的對(duì)象可以以序列化的方式存儲(chǔ)，序列化的過程是將對(duì)象轉(zhuǎn)換為二進(jìn)制字節(jié)流，本質(zhì)上可以理解為將非連續(xù)空間的鏈?zhǔn)酱鎯?chǔ)轉(zhuǎn)化為連續(xù)空間或塊存儲(chǔ)，在訪問時(shí)則需要進(jìn)行序列化的逆過程——反序列化，將字節(jié)流轉(zhuǎn)化為對(duì)象，序列化的方式可以節(jié)省存儲(chǔ)空間，但增加了存儲(chǔ)和讀取時(shí)候的計(jì)算開銷。

對(duì)于 Spark 中序列化的對(duì)象，由于是字節(jié)流的形式，其占用的內(nèi)存大小可直接計(jì)算，而對(duì)于非序列化的對(duì)象，其占用的內(nèi)存是通過周期性地采樣近似估算而得，即并不是每次新增的數(shù)據(jù)項(xiàng)都會(huì)計(jì)算一次占用的內(nèi)存大小，這種方法降低了時(shí)間開銷但是有可能誤差較大，導(dǎo)致某一時(shí)刻的實(shí)際內(nèi)存有可能遠(yuǎn)遠(yuǎn)超出預(yù)期。此外，在被 Spark 標(biāo)記為釋放的對(duì)象實(shí)例，很有可能在實(shí)際上并沒有被 JVM 回收，導(dǎo)致實(shí)際可用的內(nèi)存小于 Spark 記錄的可用內(nèi)存。所以 Spark 并不能準(zhǔn)確記錄實(shí)際可用的堆內(nèi)內(nèi)存，從而也就無法完全避免內(nèi)存溢出（OOM, Out of Memory）的異常。

雖然不能精準(zhǔn)控制堆內(nèi)內(nèi)存的申請(qǐng)和釋放，但 Spark 通過對(duì)存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存各自獨(dú)立的規(guī)劃管理，可以決定是否要在存儲(chǔ)內(nèi)存里緩存新的 RDD，以及是否為新的任務(wù)分配執(zhí)行內(nèi)存，在一定程度上可以提升內(nèi)存的利用率，減少異常的出現(xiàn)。

1.2 堆外內(nèi)存

為了進(jìn)一步優(yōu)化內(nèi)存的使用以及提高 Shuffle 時(shí)排序的效率，Spark 引入了堆外（Off-heap）內(nèi)存，使之可以直接在工作節(jié)點(diǎn)的系統(tǒng)內(nèi)存中開辟空間，存儲(chǔ)經(jīng)過序列化的二進(jìn)制數(shù)據(jù)。利用 JDK Unsafe API（從 Spark 2.0 開始，在管理堆外的存儲(chǔ)內(nèi)存時(shí)不再基于 Tachyon，而是與堆外的執(zhí)行內(nèi)存一樣，基于 JDK Unsafe API 實(shí)現(xiàn)[3]），Spark 可以直接操作系統(tǒng)堆外內(nèi)存，減少了不必要的內(nèi)存開銷，以及頻繁的 GC 掃描和回收，提升了處理性能。堆外內(nèi)存可以被精確地申請(qǐng)和釋放，而且序列化的數(shù)據(jù)占用的空間可以被精確計(jì)算，所以相比堆內(nèi)內(nèi)存來說降低了管理的難度，也降低了誤差。

在默認(rèn)情況下堆外內(nèi)存并不啟用，可通過配置 spark.memory.offHeap.enabled 參數(shù)啟用，并由 spark.memory.offHeap.size 參數(shù)設(shè)定堆外空間的大小。除了沒有 other 空間，堆外內(nèi)存與堆內(nèi)內(nèi)存的劃分方式相同，所有運(yùn)行中的并發(fā)任務(wù)共享存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存。

1.3 內(nèi)存管理接口

Spark 為存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存的管理提供了統(tǒng)一的接口——MemoryManager，同一個(gè) Executor 內(nèi)的任務(wù)都調(diào)用這個(gè)接口的方法來申請(qǐng)或釋放內(nèi)存:

清單 1 . 內(nèi)存管理接口的主要方法

 //申請(qǐng)存儲(chǔ)內(nèi)存 
def acquireStorageMemory(blockId: BlockId, numBytes: Long, memoryMode: MemoryMode): Boolean
 //申請(qǐng)展開內(nèi)存 
def acquireUnrollMemory(blockId: BlockId, numBytes: Long, memoryMode: MemoryMode): Boolean 
//申請(qǐng)執(zhí)行內(nèi)存
 def acquireExecutionMemory(numBytes: Long, taskAttemptId: Long, memoryMode: MemoryMode): Long
 //釋放存儲(chǔ)內(nèi)存 
def releaseStorageMemory(numBytes: Long, memoryMode: MemoryMode): Unit 
//釋放執(zhí)行內(nèi)存 
def releaseExecutionMemory(numBytes: Long, taskAttemptId: Long, memoryMode: MemoryMode): Unit 
//釋放展開內(nèi)存 
def releaseUnrollMemory(numBytes: Long, memoryMode: MemoryMode): Unit

 我們看到，在調(diào)用這些方法時(shí)都需要指定其內(nèi)存模式（MemoryMode），這個(gè)參數(shù)決定了是在堆內(nèi)還是堆外完成這次操作。

MemoryManager 的具體實(shí)現(xiàn)上，Spark 1.6 之后默認(rèn)為統(tǒng)一管理（）方式，1.6 之前采用的靜態(tài)管理（）方式仍被保留，可通過配置 spark.memory.useLegacyMode 參數(shù)啟用。兩種方式的區(qū)別在于對(duì)空間分配的方式，下面的第 2 小節(jié)會(huì)分別對(duì)這兩種方式進(jìn)行介紹。

2 . 內(nèi)存空間分配

2.1 靜態(tài)內(nèi)存管理

在 Spark 最初采用的靜態(tài)內(nèi)存管理機(jī)制下，存儲(chǔ)內(nèi)存、執(zhí)行內(nèi)存和其他內(nèi)存的大小在 Spark 應(yīng)用程序運(yùn)行期間均為固定的，但用戶可以應(yīng)用程序啟動(dòng)前進(jìn)行配置，堆內(nèi)內(nèi)存的分配如圖 2 所示：

圖 2 . 靜態(tài)內(nèi)存管理圖示——堆內(nèi)

可以看到，可用的堆內(nèi)內(nèi)存的大小需要按照下面的方式計(jì)算：

清單 2 . 可用堆內(nèi)內(nèi)存空間

可用的存儲(chǔ)內(nèi)存 = systemMaxMemory * spark.storage.memoryFraction * 
spark.storage.safetyFraction 
可用的執(zhí)行內(nèi)存 = systemMaxMemory * spark.shuffle.memoryFraction * 
spark.shuffle.safetyFraction 

其中 systemMaxMemory 取決于當(dāng)前 JVM 堆內(nèi)內(nèi)存的大小，最后可用的執(zhí)行內(nèi)存或者存儲(chǔ)內(nèi)存要在此基礎(chǔ)上與各自的 memoryFraction 參數(shù)和 safetyFraction 參數(shù)相乘得出。上述計(jì)算公式中的兩個(gè) safetyFraction 參數(shù)，其意義在于在邏輯上預(yù)留出 1-safetyFraction 這么一塊保險(xiǎn)區(qū)域，降低因?qū)嶋H內(nèi)存超出當(dāng)前預(yù)設(shè)范圍而導(dǎo)致 OOM 的風(fēng)險(xiǎn)（上文提到，對(duì)于非序列化對(duì)象的內(nèi)存采樣估算會(huì)產(chǎn)生誤差）。值得注意的是，這個(gè)預(yù)留的保險(xiǎn)區(qū)域僅僅是一種邏輯上的規(guī)劃，在具體使用時(shí) Spark 并沒有區(qū)別對(duì)待，和”其它內(nèi)存”一樣交給了 JVM 去管理。

堆外的空間分配較為簡(jiǎn)單，只有存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存，如圖 3 所示。可用的執(zhí)行內(nèi)存和存儲(chǔ)內(nèi)存占用的空間大小直接由參數(shù) spark.memory.storageFraction 決定，由于堆外內(nèi)存占用的空間可以被精確計(jì)算，所以無需再設(shè)定保險(xiǎn)區(qū)域。

圖 3 . 靜態(tài)內(nèi)存管理圖示——堆外

靜態(tài)內(nèi)存管理機(jī)制實(shí)現(xiàn)起來較為簡(jiǎn)單，但如果用戶不熟悉 Spark 的存儲(chǔ)機(jī)制，或沒有根據(jù)具體的數(shù)據(jù)規(guī)模和計(jì)算任務(wù)或做相應(yīng)的配置，很容易造成”一半海水，一半火焰”的局面，即存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存中的一方剩余大量的空間，而另一方卻早早被占滿，不得不淘汰或移出舊的內(nèi)容以存儲(chǔ)新的內(nèi)容。由于新的內(nèi)存管理機(jī)制的出現(xiàn)，這種方式目前已經(jīng)很少有開發(fā)者使用，出于兼容舊版本的應(yīng)用程序的目的，Spark 仍然保留了它的實(shí)現(xiàn)。

2.2 統(tǒng)一內(nèi)存管理

Spark 1.6 之后引入的統(tǒng)一內(nèi)存管理機(jī)制，與靜態(tài)內(nèi)存管理的區(qū)別在于存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存共享同一塊空間，可以動(dòng)態(tài)占用對(duì)方的空閑區(qū)域，如圖 4 和圖 5 所示

圖 4 . 統(tǒng)一內(nèi)存管理圖示——堆內(nèi)

圖 5 . 統(tǒng)一內(nèi)存管理圖示——堆外

其中最重要的優(yōu)化在于動(dòng)態(tài)占用機(jī)制，其規(guī)則如下：

• 設(shè)定基本的存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存區(qū)域（spark.storage.storageFraction 參數(shù)），該設(shè)定確定了雙方各自擁有的空間的范圍 
• 雙方的空間都不足時(shí)，則存儲(chǔ)到硬盤；若己方空間不足而對(duì)方空余時(shí)，可借用對(duì)方的空間;（存儲(chǔ)空間不足是指不足以放下一個(gè)完整的 Block）
• 執(zhí)行內(nèi)存的空間被對(duì)方占用后，可讓對(duì)方將占用的部分轉(zhuǎn)存到硬盤，然后”歸還”借用的空間
• 存儲(chǔ)內(nèi)存的空間被對(duì)方占用后，無法讓對(duì)方”歸還”，因?yàn)樾枰紤] Shuffle 過程中的很多因素，實(shí)現(xiàn)起來較為復(fù)雜

圖 6 . 動(dòng)態(tài)占用機(jī)制圖示

憑借統(tǒng)一內(nèi)存管理機(jī)制，Spark 在一定程度上提高了堆內(nèi)和堆外內(nèi)存資源的利用率，降低了開發(fā)者維護(hù) Spark 內(nèi)存的難度，但并不意味著開發(fā)者可以高枕無憂。譬如，所以如果存儲(chǔ)內(nèi)存的空間太大或者說緩存的數(shù)據(jù)過多，反而會(huì)導(dǎo)致頻繁的全量垃圾回收，降低任務(wù)執(zhí)行時(shí)的性能，因?yàn)榫彺娴?RDD 數(shù)據(jù)通常都是長(zhǎng)期駐留內(nèi)存的 。所以要想充分發(fā)揮 Spark 的性能，需要開發(fā)者進(jìn)一步了解存儲(chǔ)內(nèi)存和執(zhí)行內(nèi)存各自的管理方式和實(shí)現(xiàn)原理。

未完待續(xù)......