本篇文章主要剖析BlockManager相關的類以及總結Spark底層儲存體系。

總述

先看 BlockManager相關類之間的關係如下：

我們從NettyRpcEnv 開始，做一下簡單說明。

NettyRpcEnv是Spark 的預設的RpcEnv實現，它提供了個Spark 叢集各個節點的底層通訊環境，可以參照文章 spark 原始碼分析之十二--Spark RPC剖析之Spark RPC總結 做深入瞭解。

MemoryManager 主要負責Spark記憶體管理，可以參照 spark 原始碼分析之十五 -- Spark記憶體管理剖析做深入瞭解。

MemoryStore 主要負責Spark單節點的記憶體儲存，可以參照 spark 原始碼分析之十六 -- Spark記憶體儲存剖析 做深入瞭解。

DiskStore 主要負責Spark單節點的磁碟儲存，可以參照 spark 原始碼分析之十七 -- Spark磁碟儲存剖析 做深入瞭解。

SecurityManager 主要負責底層通訊的安全認證。

BlockManagerMaster 主要負責在executor端和driver的通訊，封裝了 driver的RpcEndpointRef。

NettyBlockTransferService 使用netty來獲取一組資料塊。

MapOutputTracker 是一個跟蹤 stage 的map 輸出位置的類，driver 和 executor 有對應的實現，分別是 MapOutputTrackerMaster 和 MapOutputTrackerWorker。

ShuffleManager在SparkEnv中初始化，它在driver端和executor端都有，負責driver端生成shuffle以及executor的資料讀寫。

BlockManager 是Spark儲存體系裡面的核心類，它執行在每一個節點上（drievr或executor），提供寫或讀本地或遠端的block到各種各樣的儲存介質中，包括磁碟、堆內記憶體、堆外記憶體。 

下面我們剖析一下之前沒有剖析過，圖中有的類：

SecurityManager

概述

類說明如下：

Spark class responsible for security. In general this class should be instantiated by the SparkEnv and most components should access it from that. 
There are some cases where the SparkEnv hasn't been initialized yet and this class must be instantiated directly. 
This class implements all of the configuration related to security features described in the "Security" document. 
Please refer to that document for specific features implemented here.

這個類主要就是負責Spark的安全的。它是由SparkEnv初始化的。

類結構

其結構如下：

成員變數

WILDCARD_ACL：常量為*，表示允許所有的組或使用者擁有檢視或修改的許可權。

authOn：表示網路傳輸是否啟用安全，由引數 spark.authenticate控制，預設為 false。

aclsOn：表示，由引數 spark.acls.enable 或 spark.ui.acls.enable 控制，預設為 false。

adminAcls：管理員許可權，由 spark.admin.acls 引數控制，預設為空字串。

adminAclsGroups：管理員所在組許可權，由 spark.admin.acls.groups 引數控制，預設為空字串。

viewAcls：檢視控制訪問列表使用者。

viewAclsGroups：檢視控制訪問列表使用者組。

modifyAcls：修改控制訪問列表使用者。

modifyAclsGroups：修改控制訪問列表使用者組。

defaultAclUsers：預設控制訪問列表使用者。由user.name 引數和 SPARK_USER環境變數一起設定。

secretKey：安全金鑰。

hadoopConf：hadoop的配置物件。

defaultSSLOptions：預設安全選項，如下：

其中SSLOption的parse 方法如下，主要用於一些安全配置的載入：

defaultSSLOptions跟getSSLOptions方法搭配使用：

核心方法

1. 設定獲取 adminAcls、viewAclsGroups、modifyAcls、modifyAclsGroups變數的方法，比較簡單，不再說明。

2. 檢查UI檢視的許可權以及修改許可權：

3. 獲取安全金鑰：

4. 獲取安全使用者：

5. 初始化安全：

總結

這個類主要是用於Spark安全的，主要包含了許可權的設定和獲取的方法，金鑰的獲取、安全使用者的獲取、許可權驗證等功能。

下面來看一下BlockManagerMaster類。

BlockManagerMaster

概述

BlockManagerMaster 這個類是對 driver的 EndpointRef 的包裝，可以說是 driver EndpointRef的一個代理類，在請求訪問driver的時候，呼叫driver的EndpointRef的對應方法，並處理其返回。

類結構 

其類結構如下：

主要是一些通過driver獲取的節點或block、或BlockManager資訊的功能函式。

成員變數

driverEndpoint是一個EndpointRef 物件，可以指本地的driver 的endpoint 或者是遠端的 endpoint引用，通過它既可以和本地的driver進行通訊，也可以和遠端的driver endpoint 進行通訊。

timeout 是指的 Spark RPC 超時時間，預設為 120s，可以通過spark.rpc.askTimeout 或 spark.network.timeout 引數來設定。

核心方法：

1. 移除executor，有同步和非同步兩種方案，這兩個方法只會在driver端使用。如下：

2. 向driver註冊blockmanager

3. 更新block資訊

4. 向driver請求獲取block對應的 location資訊

5. 向driver 請求獲得叢集中所有的 blockManager的資訊

4. 向driver 請求executor endpoint ref 物件

5. 移除block、RDD、shuffle、broadcast

6. 向driver 請求獲取每一個BlockManager記憶體狀態

7. 向driver請求獲取磁碟狀態

8. 向driver請求獲取block狀態

9. 是否有匹配的block

10.檢查是否快取了block

其依賴方法tell 方法如下：

總結

BlockManagerMaster 主要負責和driver的互動，來獲取跟底層儲存相關的資訊。

ShuffleClient

類說明

它定義了從executor或者是外部服務讀取shuffle資料的介面。

核心方法

1. init方法用於初始化ShuffleClient，需要指定executor 的appId

2. fetchBlocks 用於非同步從另一個節點請求獲取blocks，引數解釋如下：

host – the host of the remote node.
port – the port of the remote node.
execId – the executor id.
blockIds – block ids to fetch.
listener – the listener to receive block fetching status.
downloadFileManager – DownloadFileManager to create and clean temp files. If it's not null, the remote blocks will be streamed into temp shuffle files to reduce the memory usage, otherwise, they will be kept in memory.

3. shuffleMetrics 用於記錄shuffle相關的metrics資訊

BlockTransferService

類說明

它是ShuffleClient的子類。它是ShuffleClient的抽象實現類，定義了讀取shuffle的基礎框架。

核心方法

init 方法，它額外提供了使用BlockDataManager初始化的方法，方便從本地獲取block或者將block存入本地。

close：關閉ShuffleClient

port：服務正在監聽的埠

hostname：服務正在監聽的hostname

fetchBlocks 跟繼承類一樣，沒有實現，由於繼承關係可以不寫。

uploadBlocks：上傳block到遠端節點，返回一個future物件

fetchBlockSync：同步抓取遠端節點的block，直到block資料獲取成功才返回，如下：

它定義了block 抓取後，對返回結果處理的基本框架。

 uploadBlockSync 方法：同步上傳資訊，直到上傳成功才結束。如下：

ManagedBuffer的三個子類

在 spark 原始碼分析之十七 -- Spark磁碟儲存剖析 中已經提及過ManagedBuffer類。

下面看一下ManagedBuffler的三個子類：FileSegmentManagedBuffer、EncryptedManagedBuffer、NioManagedBuffer

FileSegmentManagedBuffer：由檔案中的段支援的ManagedBuffer。

EncryptedManagedBuffer：由加密檔案中的段支援的ManagedBuffer。

NioManagedBuffer：由ByteBuffer支援的ManagedBuffer。

NettyBlockTransferService

類說明：

它是BlockTransferService，使用netty來一次性獲取shuffle的block資料。

成員變數

hostname：TransportServer 監聽的hostname

serializer：JavaSerializer 例項，用於序列化反序列化java物件。

authEnabled：是否啟用安全

transportConf：TransportConf 物件，主要是用於初始化shuffle的執行緒數等配置。，spark.shuffle.io.serverThreads 和 spark.shuffle.io.clientThreads，預設是執行緒數在 [1,8] 個，這跟可用core的數量和指定core數量有關。 這兩個引數決定了底層netty server端和client 端的執行緒數。

transportContext：TransportContext 用於建立TransportServer和TransportClient的上下文。

server：TransportServer物件，是Netty的server端執行緒。

clientFactory：TransportClientFactory 用於建立TransportClient

appId：application id，由 spark.app.id 引數指定

核心方法

1. init 方法主要用於初始化底層netty的server和client，如下：

關於底層RPC部分的內容，在Spark RPC 剖析系列已經做過說明，參照 spark 原始碼分析之十二--Spark RPC剖析之Spark RPC總結 做進一步瞭解。

2. 關閉ShuffleClient：

3. 上傳資料：

config.MAX_REMOTE_BLOCK_SIZE_FETCH_TO_MEM 是由spark.maxRemoteBlockSizeFetchToMem引數決定的，預設是 整數最大值 - 512.

所以整數範圍內的block資料，是由 netty RPC來處理的,128MB顯然是在整數範圍內的，所以hdfs上的block 資料spark都是通過netty rpc來通訊傳輸的。

4. 從遠端節點獲取block資料，原始碼如下：

首先資料抓取是可以支援重試的，重試次數預設是3次，可以由引數 spark.shuffle.io.maxRetries 指定，實際上是由OneForOneBlockFetcher來遠端抓取資料的。

重試抓取遠端block機制的設計

當重試次數不大於0時，直接使用的是BlockFetchStarter來生成 OneForOneBlockFetcher 抓取資料。

當次數大於0 時，則使用 RetryingBlockFetcher 來重試式抓取資料。

先來看一下其成員變數：

executorService： 用於等待執行重試任務的共享執行緒池

fetchStarter：初始化 OneForOneBlockFetcher 物件

listener：監聽抓取block成功或失敗的listener

maxRetries；最大重試次數。

retryWaitTime：下一次重試間隔時間。可以通過 spark.shuffle.io.retryWait引數設定，預設是 5s。

retryCount：已重試次數。

outstandingBlocksIds：剩餘需要抓取的blockId集合。

currentListener：它只監聽當前fetcher的返回。

核心方法：

思路：首先，初始化需要抓取的blockId列表，已重試次數，以及currentListener。然後去呼叫fetcherStarter開始抓取任務，每一個block抓取成功後，都會呼叫currentListener對應成功方法，失敗則會呼叫 currentListener 失敗方法。在fetch過程中資料有異常出現，則先判斷是否需要重試，若需重試，則初始化重試，將wait和fetch任務放到共享執行緒池中去執行。

下面看一下，相關方法和類：

1. RetryingBlockFetchListener 類。它有兩個方法，一個是抓取成功的回撥，一個是抓取失敗的回撥。

在抓取成功回撥中，會先判斷當前的currentListener是否是它本身，並且返回的blockId在需要抓取的blockId列表中，若兩個條件都滿足，則會從需要抓取的blockId列表中把該blockId移除並且去呼叫listener相對應的抓取成功方法。

在抓取失敗回撥中，會先判斷當前的currentListener是否是它本身，並且返回的blockId在需要抓取的blockId列表中，若兩個條件都滿足，再判斷是否需要重試，如需重試則重置重試機制，否則直接呼叫listener的抓取失敗方法。

2. 是否需要重試：

思路：如果是IO 異常並且還有剩餘重試次數，則重試。

3.  初始化重試：

總結：該重試的blockFetcher 引入了中間層，即自定義的RetryingBlockFetchListener 監聽器，來完成重試或事件的傳播機制（即呼叫原來的監聽器的抓取失敗成功對應方法）以及需要抓取的blockId列表的更新，重試次數的更新等操作。

MapOutputTracker

類說明

MapOutputTracker 是一個定位跟蹤 stage 的map 輸出位置的類，driver 和 executor 有對應的實現，分別是 MapOutputTrackerMaster 和 MapOutputTrackerWorker。

其類結構如下：

成員變數

trackerEndpoint：它是一個EndpointRef物件，是driver端 MapOutputTrackerMasterEndpoint 的在executor的代理物件。

epoch：The driver-side counter is incremented every time that a map output is lost. This value is sent to executors as part of tasks, where executors compare the new epoch number to the highest epoch number that they received in the past. If the new epoch number is higher then executors will clear their local caches of map output statuses and will re-fetch (possibly updated) statuses from the driver.

eposhLock： 一個鎖物件

核心方法

1. 向driver端trackerEndpoint 傳送訊息

2. excutor 獲取每一個shuffle中task 需要讀取的範圍的 block資訊，partition範圍包頭不包尾。

3. 刪除指定的shuffle的狀態資訊

4. 停止服務

其子類MapOutputTrackerMaster 和 MapOutputTrackerWorker在後續shuffle 剖許再作進一步說明。 

ShuffleManager

類說明

它是一個可插拔的shuffle系統，ShuffleManager 在driver和每一個executor的SparkEnv中基於spark.shuffle.manager引數建立，driver使用這個類來註冊shuffle，executor或driver本地任務可以請求ShuffleManager 來讀寫任務。

類結構

1. registerShuffle：Register a shuffle with the manager and obtain a handle for it to pass to tasks.

2. getWriter：Get a writer for a given partition. Called on executors by map tasks.

3. getReader：Get a reader for a range of reduce partitions (startPartition to endPartition-1, inclusive). Called on executors by reduce tasks.

4. unregisterShuffle：Remove a shuffle's metadata from the ShuffleManager.

5. shuffleBlockResolver：Return a resolver capable of retrieving shuffle block data based on block coordinates.

6. stop：Shut down this ShuffleManager.

其有唯一子類 SortShuffleManager，我們在剖析spark shuffle 過程時，再做進一步說明。 

下面，我們來看Spark儲存體系裡面的重頭戲 -- BlockManager

BlockManager

類說明

Manager running on every node (driver and executors) which provides interfaces for putting and retrieving blocks both locally and remotely into various stores (memory, disk, and off-heap). 
Note that initialize( ) must be called before the BlockManager is usable.

它執行在每一個節點上（drievr或executor），提供寫或讀本地或遠端的block到各種各樣的儲存介質中，包括磁碟、堆內記憶體、堆外記憶體。

構造方法

其中涉及的變數，之前基本上都已作說明，不再說明。

這個類結構非常龐大，不再展示類結構圖。下面分別對其成員變數和比較重要的方法做一下說明。

成員變數

externalShuffleServiceEnabled： 是否啟用外部shuffle 服務，通過spark.shuffle.service.enabled 引數配置，預設是false

remoteReadNioBufferConversion：是否 xxxxx， 通過 spark.network.remoteReadNioBufferConversion 引數配置，預設是 false

diskBlockManager：DiskBlockManager物件，用於管理block和物理block檔案的對映關係的

blockInfoManager：BlockInfoManager物件，Block讀寫鎖

futureExecutionContext：ExecutionContextExecutorService 內部封裝了一個執行緒池，執行緒字首為 block-manager-future，最大執行緒數是 128

memoryStore：MemoryStore 物件，用於記憶體儲存。

diskStore：DiskStore物件，用於磁碟儲存。

maxOnHeapMemory：最大堆內記憶體

maxOffHeapMemory：最大堆外記憶體

externalShuffleServicePort： 外部shuffle 服務埠，通過 spark.shuffle.service.port 引數設定，預設為 7337

blockManagerId：BlockManagerId 物件是blockManager的唯一標識

shuffleServerId：BlockManagerId 物件，提供shuffle服務的BlockManager的唯一標識

shuffleClient：如果啟用了外部儲存，即externalShuffleServiceEnabled為true，使用ExternalShuffleClient，否則使用通過構造引數傳過來的 blockTransferService 物件。

maxFailuresBeforeLocationRefresh：下次從driver重新整理block location時需要重試的最大次數。通過spark.block.failures.beforeLocationRefresh 引數來設定，預設時 5

slaveEndpoint：BlockManagerSlaveEndpoint的ref物件，負責監聽處理master的請求。

asyncReregisterTask：非同步註冊任務

asyncReregisterLock：鎖物件

cachedPeers：Spark叢集中所有的BlockManager

peerFetchLock：鎖物件，用於獲取spark 叢集中所有的blockManager時用

lastPeerFetchTime：最近獲取spark 叢集中所有blockManager的時間

blockReplicationPolicy：BlockReplicationPolicy 物件，它有兩個子類 BasicBlockReplicationPolicy 和 RandomBlockReplicationPolicy。

remoteBlockTempFileManager：RemoteBlockDownloadFileManager 物件

maxRemoteBlockToMem：通過 spark.maxRemoteBlockSizeFetchToMem 引數控制，預設為整數最大值 - 512

核心方法[簡版]

注：未做過多的分析，大部分內容在之前記憶體儲存和磁碟儲存中都已涉及。

1. 初始化方法

思路：初始化 blockReplicationPolicy， 可以通過引數 spark.storage.replication.policy  來指定，預設為 RandomBlockReplicationPolicy；初始化BlockManagerId並想driver註冊該BlockManager；初始化shuffleServerId

2. 重新想driver註冊blockManager方法：

思路： 通過 BlockManagerMaster 想driver 註冊 BlockManager

3. 獲取block資料，如下：

其依賴方法 getLocalBytes 如下，思路：如果是shuffle的資料，則通過shuffleBlockResolver獲取block資訊，否則使用BlockInfoManager加讀鎖後，獲取資料。

doGetLocalBytes 方法如下，思路：按照是否需要反序列化、是否儲存在磁碟中，做相應處理，操作直接依賴與MemoryStore和DiskStore。

4. 儲存block資料，直接呼叫putBytes 方法：

其依賴方法如下，直接呼叫doPutBytes 方法：

doPutBytes 方法如下：

doPut 方法如下，思路，加寫鎖，執行putBody方法：

5. 儲存序列化之後的位元組資料

6. 儲存java物件：

7. 快取讀取的資料在記憶體中：

8. 獲取Saprk 叢集中其他的BlockManager資訊：

9. 同步block到其他的replicas：

其依賴方法如下：

10.把block從記憶體中驅逐：

11. 移除block：

12. 停止方法

BlockManager 主要提供寫或讀本地或遠端的block到各種各樣的儲存介質中，包括磁碟、堆內記憶體、堆外記憶體。獲取Spark 叢集的BlockManager的資訊、驅逐記憶體中block等等方法。

其遠端互動依賴於底層的netty模組。有很多的關於儲存的方法都依賴於MemoryStore和DiskStore的實現，不再做一一解釋。

總結

本篇文章介紹了Spark儲存體系的最後部分內容。行文有些倉促，有一些類可能會漏掉，但對於理解Spark 儲存體系已經綽綽有餘。本地儲存依賴於MemoryStore和DiskStore，遠端呼叫依賴於NettyBlockTransferService、BlockManagerMaster、MapOutputTracker等，其底層絕大多數依賴於netty與driver或其他executor通訊。

Spark shuffle、broadcast等也是依賴於儲存系統的。接下來將進入spark的核心部分，去探索Spark底層的RDD是如何構建Stage作業以及每一個作業是如何工作的。