本篇文章主要剖析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作業以及每一個作業是如何工作的。

?

轉載于:https://www.cnblogs.com/johnny666888/p/11226984.html

總結

以上是生活随笔為你收集整理的spark 源码分析之十八 -- Spark存储体系剖析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。