上篇spark 原始碼分析之十五 -- Spark記憶體管理剖析 講解了Spark的記憶體管理機制，主要是MemoryManager的內容。跟Spark的記憶體管理機制最密切相關的就是記憶體儲存，本篇文章主要介紹Spark記憶體儲存。

總述

跟記憶體儲存的相關類的關係如下：

MemoryStore是負責記憶體儲存的類，其依賴於BlockManager、SerializerManager、BlockInfoManager、MemoryManager。

BlockManager是BlockEvictionHandler的實現類，負責實現dropFromMemory方法，必要時從記憶體中把block丟掉，可能會轉儲到磁碟上。

SerializerManager是負責持久化的一個類，可以參考文章spark 原始碼分析之十三 -- SerializerManager剖析做深入瞭解。

BlockInfoManager是一個實現了對block讀寫時的一個鎖機制，具體可以看下文。

MemoryManager 是一個記憶體管理器，從Spark 1.6 以後，其儲存記憶體池大小和執行記憶體池大小是可以動態擴充套件的。即儲存記憶體和執行記憶體必要時可以從對方記憶體池借用空閒記憶體來滿足自己的使用需求。可以參考文章 spark 原始碼分析之十五 -- Spark記憶體管理剖析 做深入瞭解。

BlockInfo 儲存了跟block相關的資訊。

BlockId的name不同的型別有不同的格式，代表不同的block型別。

StorageLevel 表示block的儲存級別，它本身是支援序列化的。

當儲存一個集合為序列化位元組陣列時，失敗的結果由 PartiallySerializedBlock 返回。

當儲存一個集合為Java物件陣列時，失敗的結果由 PartiallyUnrolledIterator 返回。

RedirectableOutputStream 是對另一個outputstream的包裝outputstream，負責直接將資料中轉到另一個outputstream中。

ValueHolder是一個記憶體中轉站，其有一個getBuilder方法可以獲取到MemoryEntryBuilder物件，該物件會負責將中轉站的資料轉換為對應的可以儲存到MemStore中的MemoryEntry。

我們逐個來分析其原始碼：

BlockInfo

它記錄了block 的相關資訊。

level： StorageLevel 型別，代表block的儲存級別

classTag：block的對應類，用於選擇序列化類

tellMaster：block 的變化是否告知master。大部分情況下都是需要告知的，除了廣播的block。

size： block的大小（in byte）

readerCount：block 讀的次數

writerTask：當前持有該block寫鎖的taskAttemptId，其中 BlockInfo.NON_TASK_WRITER 表示非 task任務 持有鎖，比如driver執行緒，BlockInfo.NO_WRITER 表示沒有任何程式碼持有寫鎖。

BlockId

A Block can be uniquely identified by its filename, but each type of Block has a different set of keys which produce its unique name. If your BlockId should be serializable, be sure to add it to the BlockId.apply() method.

其子類，在上圖中已經標明。

BlockInfoManager

文件介紹如下：

Component of the BlockManager which tracks metadata for blocks and manages block locking. The locking interface exposed by this class is readers-writer lock. Every lock acquisition is automatically associated with a running task and locks are automatically released upon task completion or failure. This class is thread-safe.

它有三個成員變數，如下：

infos 儲存了 Block-id 和 block資訊的對應關係。

writeLocksByTask 儲存了每一個任務和任務持有寫鎖的block-id

readLockByTasks 儲存了每一個任務和任務持有讀鎖的block-id，因為讀鎖是可重入的，所以 ConcurrentHashMultiset 是支援多個重複值的。

方法如下：

1. 註冊task

2. 獲取當前task

3. 獲取讀鎖

思路：如果block存在，並且沒有task在寫，則直接讀即可，否則進入鎖等待區等待。

4. 獲取寫鎖

思路：如果block存在，且沒有task在讀，也沒有task在寫，則在寫鎖map上記錄task，表示已獲取寫鎖，否則進入等待區等待

5. 斷言有task持有寫鎖寫block

6. 寫鎖降級

思路：首先把和block繫結的task取出並和當前task比較，若是同一個task，則呼叫unlock方法

7. 釋放鎖：

思路：若當前任務持有寫鎖，則直接釋放，否則讀取次數減1，並且從讀鎖記錄中刪除一條讀鎖記錄。最後喚醒在鎖等待區等待的task。

8. 獲取為寫一個新的block獲取寫鎖

9. 釋放掉指定task的所有鎖

思路：先獲取該task的讀寫鎖記錄，然後移除寫鎖記錄集中的每一條記錄，移除讀鎖記錄集中的每一條讀鎖記錄。

10. 移除並釋放寫鎖

讀寫鎖記錄清零，解除block-id和block資訊的繫結。

還有一些查詢方法，不再做詳細說明。

簡單總結一下：

讀鎖支援可重入，即可以重複獲取讀鎖。可以獲取讀鎖的條件是：沒有task在寫該block，對有沒有task在讀block沒有要求。

寫鎖當且僅當一個task獲取，可以獲取寫鎖的條件是：沒有task在讀block，沒有task在寫block。

注意，這種設計可以用在一個block的讀的次數遠大於寫的次數的情況下。我們可以來做個假設：假設一個block寫的次數遠超過讀的次數，同時多個task寫同一個block的操作就變成了序列的，寫的效率，因為只有一個BlockInfoManager物件，即一個鎖，即所有在鎖等待區等待的writer們都在競爭一個鎖。對於讀的次數遠超過寫的次數情況下，reader們可以肆無忌憚地讀取資料資料，基本處於無鎖情況下，幾乎沒有了鎖切換帶來的開銷，並且可以允許不同task同時讀取同一個block的資料，讀的吞吐量也提高了。

總之，BlockInfoManager自己實現了block的一套讀寫鎖機制，這種讀寫鎖的設計思路是非常經典和值得學習的。 

RedirectableOutputStream

文件說明：

A wrapper which allows an open [[OutputStream]] to be redirected to a different sink.

即這個類可以將outputstream重定向到另一個outputstream。

原始碼也很簡單：

os成員變數就是重定向的目標outputstream

MemoryEntry

memoryEntry本質上就是記憶體中一個block，指向了儲存在記憶體中的真實資料。

如上圖，它有兩個子類：

其中，DeserializedMemoryEntry 是用來儲存反序列化之後的java物件陣列的，value是一個數據，儲存著真實的反序列化資料，size表示，classTag記錄著陣列中被擦除的資料的Class型別，這種資料只能儲存在堆內記憶體中。

SerializedMemoryEntry 是用來儲存序列化之後的ByteBuffer陣列的，buffer中記錄的是真實的Array[ByteBuffer]資料。memoryMode表示資料儲存的記憶體區域，堆外記憶體還是堆內記憶體，classTag記錄著序列化前被擦除的資料的Class型別，size表示位元組資料大小。

MemoryEntryBuilder

build方法將記憶體資料構建到MemoryEntry中

ValuesHolder

本質上來說，就是一個記憶體中轉站。資料被臨時寫入到這個中轉站，然後呼叫其getBuilder方法獲取 MemoryEntryBuilder 物件，這個物件用於構建MemoryEntry 物件。

storeValues用於寫入資料，estimateSize用於評估holder中記憶體的大小。呼叫getBuilder之後會返回 MemoryEntryBuilder物件，後續可以拿這個builder建立MemoryEntry

呼叫getBuilder之後，會關閉流，禁止資料寫入。

它有兩個子類：用於中轉Java物件的DeserializedValuesHolder和用於中轉位元組資料的SerializedValuesHolder。

其實現類具體如下：

1. DeserializedValuesHolder

2. SerializedValuesHolder

接下來，我們看一下Spark記憶體儲存中的重頭戲 -- MemoryStore

MemoryStore

文件說明：

Stores blocks in memory, either as Arrays of deserialized Java objects or as serialized ByteBuffers.

類內部結構如下：

對成員變數的說明：

entries 本質上就是在記憶體中儲存blockId和block內容的一個map，它的 accessOrder為true，即最近訪問的會被移動到連結串列尾部。

onHeapUnrollMemoryMap 記錄了taskAttemptId和需要攤開一個block需要的堆內記憶體大小的關係

offHeapUnrollMemoryMap 記錄了taskAttemptId和需要攤開一個block需要的堆外記憶體大小的關係

unrollMemoryThreshold 表示在攤開一個block 之前給request分配的初始記憶體，可以通過 spark.storage.unrollMemoryThreshold 來調整，預設是 1MB

下面，開門見山，直接剖析比較重要的方法：

1. putBytes：這個方法只被BlockManager呼叫，其中_bytes回撥用於生成直接被快取的ChunkedByteBuffer：

思路：先從MemoryManager中申請記憶體，如果申請成功，則呼叫回撥方法 _bytes 獲取ChunkedByteBuffer資料，然後封裝成 SerializedMemoryEntry物件 ，最後將封裝好的SerializedMemoryEntry物件快取到 entries中。

2. 把迭代器中值儲存為記憶體中的Java物件

思路：轉換為DeserializedValueHolder物件，進而呼叫putIterator方法，ValueHolder就是一個抽象，使得putIterator既可以快取序列化的位元組資料又可以快取Java物件陣列。

3. 把迭代器中值儲存為記憶體中的序列化位元組資料

思路：轉換為 SerializedValueHolder 物件，進而呼叫putIterator方法。

MAX_ROUND_ARRARY_LENGTH和unrollMemoryThreshold的定義如下：

1 public static int MAX_ROUNDED_ARRAY_LENGTH = Integer.MAX_VALUE - 15;
2 private val unrollMemoryThreshold: Long = conf.getLong("spark.storage.unrollMemoryThreshold", 1024 * 1024)

unrollMemoryThreshold 預設是 1MB，可以通過 spark.storage.unrollMemoryThreshold 引數調整大小。

4. putIterator方法由引數ValueHolder，使得快取位元組資料和Java物件可以放到一個方法來。 方法2跟3 都呼叫了 putIterator 方法，如下：

思路：

第一步：定義攤開記憶體初始化大小，攤開記憶體增長率，攤開記憶體檢查頻率等變數。

第二步：向MemoryManager請求申請攤開初始記憶體，若成功，則記錄這筆攤開記憶體。

第三步：然後進入223～240行的while迴圈，在這個迴圈裡：

• 迴圈條件：如果還有值需要攤開並且上次記憶體申請是成功的，則繼續進行該次迴圈
• 不斷想ValueHolder中add資料。如果攤開的元素個數不是UNROLL_MEMORY_CHECK_PERIOD的整數倍，則攤開個數加1；否則，檢視ValueHolder中的記憶體是否大於了已分配記憶體，若大於，則請求MemoryManager分配記憶體，並將分配的記憶體累加到已分配記憶體中。

第四步：

若上一次向MemoryManager申請記憶體成功，則從ValueHolder中獲取builder，並且計算準確記憶體開銷。查看準確記憶體是否大於了已分配記憶體，若大於，則請求MemoryManager分配記憶體，並將分配的記憶體累加到已分配記憶體中。

否則，否則列印記憶體使用情況，返回為攤開該block申請的記憶體

第五步：

若上一次向MemoryManager申請記憶體成功，首先呼叫MemoryEntryBuilder的build方法構建出可以直接存入記憶體的MemoryEntry，並向MemoryManager請求釋放攤開記憶體，申請儲存記憶體，並確保儲存記憶體申請成功。最後將資料存入記憶體的entries中。 

否則列印記憶體使用情況，返回為攤開該block申請的記憶體

其實之前不是很理解unroll這個詞在這裡的含義，一直譯作攤開，它其實指的就是集合的資料轉儲到中轉站這個操作，攤開記憶體指這個操作需要的記憶體。

下面來看一下這個方法裡面依賴的常量和方法： 

4. 1 unrollMemoryThreshold 在上一個方法已做說明。UNROLL_MEMORY_CHECK_PERIOD 和 UNROLL_MEMORY_GROWTH_FACTOR 常量定義如下：

即，UNROLL_MEMORY_CHECK_PERIOD預設是16，UNROLL_MEMORY_GROWTH_FACTOR 預設是 1.5

4.2 reserveUnrollMemoryForThisTask方法原始碼如下，思路大致上是先從MemoryManager 申請攤開記憶體，若成功，則根據memoryMode在堆內或堆外記錄攤開記憶體的map上記錄新分配的記憶體。

4.3 releaseUnrollMemoryForThisTask方法如下，實現思路：先根據memoryMode獲取到對應記錄堆內或堆外記憶體的使用情況的map，然後在該task的攤開記憶體上減去這筆記憶體開銷，如果減完之後，task使用記憶體為0，則直接從map中移除對該task的記憶體記錄。

4.4 日誌列印block攤開記憶體和當前記憶體使用情況

5. 獲取快取的值：

思路：直接根據blockId從entries中取出MemoryEntry資料，然後根據MemoryEntry型別取出資料即可。

6. 移除Block或清除快取，比較簡單，不做過多說明：

7. 嘗試驅逐block來釋放指定大小的記憶體空間來儲存給定的block，程式碼如下：

該方法有三個引數：要分配記憶體的blockId，block大小，記憶體型別(堆內還是堆外)。

第 469～485 行：dropBlock 方法思路： 先從MemoryEntry中獲取data，再呼叫 BlockManager從記憶體中驅逐出該block，如果該block 的StorageLevel允許落地到磁碟，則先落到磁碟，再從記憶體中刪除之，最後更新該block的StorageLevel，最後檢查新的StorageLevel，若該block還在記憶體或磁碟中，則釋放鎖，否則，直接從BlockInfoManager中刪除之。

第 443 行： 找到block對應的rdd。

第451～467 行：先給entries上鎖，然後遍歷entries集合，檢查block 是否可以從記憶體中驅逐，若可以則把它加入到selectedBlocks集合中，並把該block大小累加到freedMemory中。

461行的 lockForWriting 方法，不堵塞，即如果第一次拿不到寫鎖，則一直不停地輪詢，直到可以拿到寫鎖為止。那麼問題來了，為什麼要先獲取寫鎖呢？因為寫鎖具有排他性並且不具備可重入性，一旦拿到寫鎖，其他鎖就不能再訪問該block了。

487行～ 528 行：若計劃要釋放的記憶體小於儲存新block需要的記憶體大小，則直接釋放寫鎖，不從記憶體中驅逐之前選擇的block，直接返回。

若計劃要釋放的記憶體不小於儲存新block需要的記憶體大小，則遍歷之前選擇的每一個block，獲取entry，並呼叫dropMemory方法，返回釋放的記憶體大小。finally 程式碼塊是防止在dropMemory過程中，該執行緒被中斷，其餘block寫鎖不能被釋放的情況。

其依賴的方法如下：

儲存記憶體失敗之後，會返回 PartiallySerializedBlock 或者 PartiallyUnrolledIterator。

PartiallyUnrolledIterator 是一個Iterator，可以用來遍歷block資料，同時負責釋放攤開記憶體。

PartiallySerializedBlock 它可以將失敗的block轉化成 PartiallyUnrolledIterator 用來遍歷，可以直接丟棄失敗的block，也可以把資料轉儲到給定的可以落地的outputstream中，同時釋放攤開記憶體。

總結：

本篇文章主要講解了Spark的記憶體儲存相關的內容，重點講解了BlockInfoManager實現的鎖機制、跟ValuesHolder中轉站相關的MemoryEntry、EmmoryEntryBuilder等相關內容以及記憶體儲存中的重頭戲 -- MemStore相關的Block儲存、Block釋放、為新Block驅逐記憶體等等功