3. 存儲內(nèi)存管理

3.1 RDD 的持久化機制

彈性分布式數(shù)據(jù)集（RDD）作為 Spark 最根本的數(shù)據(jù)抽象，是只讀的分區(qū)記錄（Partition）的集合，只能基于在穩(wěn)定物理存儲中的數(shù)據(jù)集上創(chuàng)建，或者在其他已有的 RDD 上執(zhí)行轉(zhuǎn)換（Transformation）操作產(chǎn)生一個新的 RDD。轉(zhuǎn)換后的 RDD 與原始的 RDD 之間產(chǎn)生的依賴關(guān)系，構(gòu)成了血統(tǒng)（Lineage）。憑借血統(tǒng)，Spark 保證了每一個 RDD 都可以被重新恢復(fù)。但 RDD 的所有轉(zhuǎn)換都是惰性的，即只有當一個返回結(jié)果給 Driver 的行動（Action）發(fā)生時，Spark 才會創(chuàng)建任務(wù)讀取 RDD，然后真正觸發(fā)轉(zhuǎn)換的執(zhí)行。

Task 在啟動之初讀取一個分區(qū)時，會先判斷這個分區(qū)是否已經(jīng)被持久化，如果沒有則需要檢查 Checkpoint 或按照血統(tǒng)重新計算。所以如果一個 RDD 上要執(zhí)行多次行動，可以在第一次行動中使用 persist 或 cache 方法，在內(nèi)存或磁盤中持久化或緩存這個 RDD，從而在后面的行動時提升計算速度。事實上，cache 方法是使用默認的 MEMORY_ONLY 的存儲級別將 RDD 持久化到內(nèi)存，故緩存是一種特殊的持久化。 堆內(nèi)和堆外存儲內(nèi)存的設(shè)計，便可以對緩存 RDD 時使用的內(nèi)存做統(tǒng)一的規(guī)劃和管 理 （存儲內(nèi)存的其他應(yīng)用場景，如緩存 broadcast 數(shù)據(jù)，暫時不在本文的討論范圍之內(nèi)）。

RDD 的持久化由 Spark 的 Storage 模塊 [7] 負責(zé)，實現(xiàn)了 RDD 與物理存儲的解耦合。Storage 模塊負責(zé)管理 Spark 在計算過程中產(chǎn)生的數(shù)據(jù)，將那些在內(nèi)存或磁盤、在本地或遠程存取數(shù)據(jù)的功能封裝了起來。在具體實現(xiàn)時 Driver 端和 Executor 端的 Storage 模塊構(gòu)成了主從式的架構(gòu)，即 Driver 端的 BlockManager 為 Master，Executor 端的 BlockManager 為 Slave。Storage 模塊在邏輯上以 Block 為基本存儲單位，RDD 的每個 Partition 經(jīng)過處理后唯一對應(yīng)一個 Block（BlockId 的格式為 rdd_RDD-ID_PARTITION-ID ）。Master 負責(zé)整個 Spark 應(yīng)用程序的 Block 的元數(shù)據(jù)信息的管理和維護，而 Slave 需要將 Block 的更新等狀態(tài)上報到 Master，同時接收 Master 的命令，例如新增或刪除一個 RDD。

圖 7 . Storage 模塊示意圖

在對 RDD 持久化時，Spark 規(guī)定了 MEMORY_ONLY、MEMORY_AND_DISK 等 7 種不同的 ，而存儲級別是以下 5 個變量的組合：

清單 3 . 存儲級別

class StorageLevel private( 
private var _useDisk: Boolean, //磁盤 
private var _useMemory: Boolean, //這里其實是指堆內(nèi)內(nèi)存 
private var _useOffHeap: Boolean, //堆外內(nèi)存 
private var _deserialized: Boolean, //是否為非序列化
private var _replication: Int = 1 //副本個數(shù) ) 

通過對數(shù)據(jù)結(jié)構(gòu)的分析，可以看出存儲級別從三個維度定義了 RDD 的 Partition（同時也就是 Block）的存儲方式：

• 存儲位置：磁盤／堆內(nèi)內(nèi)存／堆外內(nèi)存。如 MEMORY_AND_DISK 是同時在磁盤和堆內(nèi)內(nèi)存上存儲，實現(xiàn)了冗余備份。OFF_HEAP 則是只在堆外內(nèi)存存儲，目前選擇堆外內(nèi)存時不能同時存儲到其他位置。
• 存儲形式：Block 緩存到存儲內(nèi)存后，是否為非序列化的形式。如 MEMORY_ONLY 是非序列化方式存儲，OFF_HEAP 是序列化方式存儲。
• 副本數(shù)量：大于 1 時需要遠程冗余備份到其他節(jié)點。如 DISK_ONLY_2 需要遠程備份 1 個副本。

3.2 RDD 緩存的過程

RDD 在緩存到存儲內(nèi)存之前，Partition 中的數(shù)據(jù)一般以迭代器（）的數(shù)據(jù)結(jié)構(gòu)來訪問，這是 Scala 語言中一種遍歷數(shù)據(jù)集合的方法。通過 Iterator 可以獲取分區(qū)中每一條序列化或者非序列化的數(shù)據(jù)項(Record)，這些 Record 的對象實例在邏輯上占用了 JVM 堆內(nèi)內(nèi)存的 other 部分的空間，同一 Partition 的不同 Record 的空間并不連續(xù)。

RDD 在緩存到存儲內(nèi)存之后，Partition 被轉(zhuǎn)換成 Block，Record 在堆內(nèi)或堆外存儲內(nèi)存中占用一塊連續(xù)的空間。將Partition由不連續(xù)的存儲空間轉(zhuǎn)換為連續(xù)存儲空間的過程，Spark稱之為”展開”（Unroll）。Block 有序列化和非序列化兩種存儲格式，具體以哪種方式取決于該 RDD 的存儲級別。非反序列化的 Block 以一種 DeserializedMemoryEntry 的數(shù)據(jù)結(jié)構(gòu)定義，用一個數(shù)組存儲所有的 Java 對象，非序列化的 Block 則以 SerializedMemoryEntry 的數(shù)據(jù)結(jié)構(gòu)定義，用字節(jié)緩沖區(qū)（ByteBuffer）來存儲二進制數(shù)據(jù)。每個 Executor 的 Storage 模塊用一個鏈式 Map 結(jié)構(gòu)（LinkedHashMap）來管理堆內(nèi)和堆外存儲內(nèi)存中所有的 Block 對象的實例[6]，對這個 LinkedHashMap 新增和刪除間接記錄了內(nèi)存的申請和釋放。

因為不能保證存儲空間可以一次容納 Iterator 中的所有數(shù)據(jù)，當前的計算任務(wù)在 Unroll 時要向 MemoryManager 申請足夠的 Unroll 空間來臨時占位，空間不足則 Unroll 失敗，空間足夠時可以繼續(xù)進行。對于序列化的 Partition，其所需的 Unroll 空間可以直接累加計算，一次申請。而非序列化的 Partition 則要在遍歷 Record 的過程中依次申請，即每讀取一條 Record，采樣估算其所需的 Unroll 空間并進行申請，空間不足時可以中斷，釋放已占用的 Unroll 空間。如果最終 Unroll 成功，當前 Partition 所占用的 Unroll 空間被轉(zhuǎn)換為正常的緩存 RDD 的存儲空間.

 3.3 淘汰和落盤

由于同一個 Executor 的所有的計算任務(wù)共享有限的存儲內(nèi)存空間，當有新的 Block 需要緩存但是剩余空間不足且無法動態(tài)占用時，就要對 LinkedHashMap 中的舊 Block 進行淘汰（Eviction），而被淘汰的 Block 如果其存儲級別中同時包含存儲到磁盤的要求，則要對其進行落盤（Drop），否則直接刪除該 Block。

存儲內(nèi)存的淘汰規(guī)則為：

• 被淘汰的舊 Block 要與新 Block 的 MemoryMode 相同，即同屬于堆外或堆內(nèi)內(nèi)存
• 新舊 Block 不能屬于同一個 RDD，避免循環(huán)淘汰
• 舊 Block 所屬 RDD 不能處于被讀狀態(tài)，避免引發(fā)一致性問題
• 遍歷 LinkedHashMap 中 Block，按照最近最少使用（LRU）的順序淘汰，直到滿足新 Block 所需的空間。其中 LRU 是 LinkedHashMap 的特性。

落盤的流程則比較簡單，如果其存儲級別符合_useDisk 為 true 的條件，再根據(jù)其_deserialized 判斷是否是非序列化的形式，若是則對其進行序列化，最后將數(shù)據(jù)存儲到磁盤，在 Storage 模塊中更新其信息。

4. 執(zhí)行內(nèi)存管理

4.1 多任務(wù)間內(nèi)存分配

Executor 內(nèi)運行的任務(wù)同樣共享執(zhí)行內(nèi)存，Spark 用一個 HashMap 結(jié)構(gòu)保存了任務(wù)到內(nèi)存耗費的映射。每個任務(wù)可占用的執(zhí)行內(nèi)存大小的范圍為 1/2N ~ 1/N，其中 N 為當前 Executor 內(nèi)正在運行的任務(wù)的個數(shù)。每個任務(wù)在啟動之時，要向 MemoryManager 請求申請最少為 1/2N 的執(zhí)行內(nèi)存，如果不能被滿足要求則該任務(wù)被阻塞，直到有其他任務(wù)釋放了足夠的執(zhí)行內(nèi)存，該任務(wù)才可以被喚醒。

4.2 Shuffle 的內(nèi)存占用

執(zhí)行內(nèi)存主要用來存儲任務(wù)在執(zhí)行 Shuffle 時占用的內(nèi)存，Shuffle 是按照一定規(guī)則對 RDD 數(shù)據(jù)重新分區(qū)的過程，我們來看 Shuffle 的 Write 和 Read 兩階段對執(zhí)行內(nèi)存的使用：

Shuffle Write

1. 若在 map 端選擇普通的排序方式，會采用 ExternalSorter 進行外排，在內(nèi)存中存儲數(shù)據(jù)時主要占用堆內(nèi)執(zhí)行空間。
2. 若在 map 端選擇 Tungsten 的排序方式，則采用 ShuffleExternalSorter 直接對以序列化形式存儲的數(shù)據(jù)排序，在內(nèi)存中存儲數(shù)據(jù)時可以占用堆外或堆內(nèi)執(zhí)行空間，取決于用戶是否開啟了堆外內(nèi)存以及堆外執(zhí)行內(nèi)存是否足夠。

Shuffle Read

1. 在對 reduce 端的數(shù)據(jù)進行聚合時，要將數(shù)據(jù)交給 Aggregator 處理，在內(nèi)存中存儲數(shù)據(jù)時占用堆內(nèi)執(zhí)行空間。
2. 如果需要進行最終結(jié)果排序，則要將再次將數(shù)據(jù)交給 ExternalSorter 處理，占用堆內(nèi)執(zhí)行空間。

在 ExternalSorter 和 Aggregator 中，Spark 會使用一種叫 AppendOnlyMap 的哈希表在堆內(nèi)執(zhí)行內(nèi)存中存儲數(shù)據(jù)，但在 Shuffle 過程中所有數(shù)據(jù)并不能都保存到該哈希表中，當這個哈希表占用的內(nèi)存會進行周期性地采樣估算，當其大到一定程度，無法再從 MemoryManager 申請到新的執(zhí)行內(nèi)存時，Spark 就會將其全部內(nèi)容存儲到磁盤文件中，這個過程被稱為溢存(Spill)，溢存到磁盤的文件最后會被歸并(Merge)。

Shuffle Write 階段中用到的 Tungsten 是 Databricks 公司提出的對 Spark 優(yōu)化內(nèi)存和 CPU 使用的計劃[9]，解決了一些 JVM 在性能上的限制和弊端。Spark 會根據(jù) Shuffle 的情況來自動選擇是否采用 Tungsten 排序。Tungsten 采用的頁式內(nèi)存管理機制建立在 MemoryManager 之上，即 Tungsten 對執(zhí)行內(nèi)存的使用進行了一步的抽象，這樣在 Shuffle 過程中無需關(guān)心數(shù)據(jù)具體存儲在堆內(nèi)還是堆外。每個內(nèi)存頁用一個 MemoryBlock 來定義，并用 Object obj 和 long offset 這兩個變量統(tǒng)一標識一個內(nèi)存頁在系統(tǒng)內(nèi)存中的地址。堆內(nèi)的 MemoryBlock 是以 long 型數(shù)組的形式分配的內(nèi)存，其 obj 的值為是這個數(shù)組的對象引用，offset 是 long 型數(shù)組的在 JVM 中的初始偏移地址，兩者配合使用可以定位這個數(shù)組在堆內(nèi)的絕對地址；堆外的 MemoryBlock 是直接申請到的內(nèi)存塊，其 obj 為 null，offset 是這個內(nèi)存塊在系統(tǒng)內(nèi)存中的 64 位絕對地址。Spark 用 MemoryBlock 巧妙地將堆內(nèi)和堆外內(nèi)存頁統(tǒng)一抽象封裝，并用頁表(pageTable)管理每個 Task 申請到的內(nèi)存頁。

Tungsten 頁式管理下的所有內(nèi)存用 64 位的邏輯地址表示，由頁號和頁內(nèi)偏移量組成：

• 頁號：占 13 位，唯一標識一個內(nèi)存頁，Spark 在申請內(nèi)存頁之前要先申請空閑頁號。
• 頁內(nèi)偏移量：占 51 位，是在使用內(nèi)存頁存儲數(shù)據(jù)時，數(shù)據(jù)在頁內(nèi)的偏移地址。

有了統(tǒng)一的尋址方式，Spark 可以用 64 位邏輯地址的指針定位到堆內(nèi)或堆外的內(nèi)存，整個 Shuffle Write 排序的過程只需要對指針進行排序，并且無需反序列化，整個過程非常高效，對于內(nèi)存訪問效率和 CPU 使用效率帶來了明顯的提升[10]。

Spark 的存儲內(nèi)存和執(zhí)行內(nèi)存有著截然不同的管理方式：對于存儲內(nèi)存來說，Spark 用一個 LinkedHashMap 來集中管理所有的 Block，Block 由需要緩存的 RDD 的 Partition 轉(zhuǎn)化而成；而對于執(zhí)行內(nèi)存，Spark 用 AppendOnlyMap 來存儲 Shuffle 過程中的數(shù)據(jù)，在 Tungsten 排序中甚至抽象成為頁式內(nèi)存管理，開辟了全新的 JVM 內(nèi)存管理機制。

結(jié)束語

Spark 的內(nèi)存管理是一套復(fù)雜的機制，且 Spark 的版本更新比較快，因此，在理解本文的操作方法的基礎(chǔ)之上，還需要結(jié)合新版本的功能，才能達到更好的效果。