新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案

spark自己的分布式存儲系統(tǒng)-BlockManager

Spark 自己的分布式存儲系統(tǒng) - BlockManager

作者：佚名 2018-05-10 09:34:21

存儲

存儲軟件

大數(shù)據(jù)

Spark

分布式 BlockManager 是 spark 中至關(guān)重要的一個組件，在 spark的的運(yùn)行過程中到處都有 BlockManager 的身影，只有搞清楚 BlockManager 的原理和機(jī)制，你才能更加深入的理解 spark。今天我們來揭開 BlockaManager 的底層原理和設(shè)計思路，

我們提供的服務(wù)有：做網(wǎng)站、網(wǎng)站建設(shè)、微信公眾號開發(fā)、網(wǎng)站優(yōu)化、網(wǎng)站認(rèn)證、遂溪ssl等。為1000+企事業(yè)單位解決了網(wǎng)站和推廣的問題。提供周到的售前咨詢和貼心的售后服務(wù)，是有科學(xué)管理、有技術(shù)的遂溪網(wǎng)站制作公司

整體架構(gòu)

BlockManager 是 spark 中至關(guān)重要的一個組件，在 spark的的運(yùn)行過程中到處都有 BlockManager 的身影，只有搞清楚 BlockManager 的原理和機(jī)制，你才能更加深入的理解 spark。今天我們來揭開 BlockaManager 的底層原理和設(shè)計思路，

BlockManager 是一個嵌入在 spark 中的 key-value型分布式存儲系統(tǒng)，是為 spark 量身打造的，

BlockManager 在一個 spark 應(yīng)用中作為一個本地緩存運(yùn)行在所有的節(jié)點(diǎn)上，包括所有 driver 和 executor上。 BlockManager 對本地和遠(yuǎn)程提供一致的 get 和set 數(shù)據(jù)塊接口， BlockManager 本身使用不同的存儲方式來存儲這些數(shù)據(jù)，包括 memory, disk, off-heap。

上面是一個整體的架構(gòu)圖， BlockManagerMaster擁有BlockManagerMasterEndpoint 的actor和所有BlockManagerSlaveEndpoint的ref，可以通過這些引用對 slave 下達(dá)命令

executor 節(jié)點(diǎn)上的BlockManagerMaster 則擁有BlockManagerMasterEndpoint的ref和自身BlockManagerSlaveEndpoint的actor?？梢酝ㄟ^ Master的引用注冊自己。

在master 和 slave 可以正常的通信之后，就可以根據(jù)設(shè)計的交互協(xié)議進(jìn)行交互，整個分布式緩存系統(tǒng)也就運(yùn)轉(zhuǎn)起來了，

初始化

我們知道， sparkEnv 啟動的時候會啟動各個組件， BlockManager 也不例外，也是這個時候啟動的，

啟動的時候會根據(jù)自己是在 driver 還是 executor 上進(jìn)行不同的啟動過程，

  
 
 
 
   
  
  
  def registerOrLookupEndpoint(
   
  
  
          name: String, endpointCreator: => RpcEndpoint):
   
  
  
        RpcEndpointRef = {
   
  
  
        if (isDriver) {
   
  
  
          logInfo("Registering " + name)
   
  
  
          rpcEnv.setupEndpoint(name, endpointCreator)
   
  
  
        } else {
   
  
  
          RpcUtils.makeDriverRef(name, conf, rpcEnv)
   
  
  
        }
   
  
  
      }

上圖是 sparkEnv 在 master上啟動的時候，構(gòu)造了一個 BlockManagerMasterEndpoint, 然后把這個Endpoint 注冊在 rpcEnv中，同時也會啟動自己的 BlockManager

上圖是 sparkEnv 在executor上啟動的時候，通過 setupEndpointRef 方法獲取到了 BlockManagerMaster的引用 BlockManagerMasterRef，同時也會啟動自己的 BlockManager，

在 BlockManager 初始化自己的時候，會向 BlockManagerMasterEndpoint 注冊自己， BlockManagerMasterEndpoint 發(fā)送 registerBlockManager消息， BlockManagerMasterEndpoint 接受到消息，把 BlockManagerSlaveEndpoint 的引用保存在自己的 blockManagerInfo 數(shù)據(jù)結(jié)構(gòu)中以待后用。

分布式協(xié)議

下面的一個表格是 master 和 slave 接受到各種類型的消息，以及接受到消息后，做的處理。

BlockManagerMasterEndpoint 接受的消息

BlockManagerSlaveEndpoint 接受的消息

根據(jù)以上的協(xié)議，相信我們可以很清楚的猜測整個交互的流程，一般過程應(yīng)該是這樣的， slave的 BlockManager 在自己接的上存儲一個 Block, 然后把這個 BlockId 匯報到master的BlockManager , 經(jīng)過 cache, shuffle 或者 Broadcast后，別的節(jié)點(diǎn)需要上一步的Block的時候，會到 master 獲取數(shù)據(jù)所在位置，然后去相應(yīng)節(jié)點(diǎn)上去 fetch。

存儲層

在RDD層面上我們了解到RDD是由不同的partition組成的，我們所進(jìn)行的transformation和action是在partition上面進(jìn)行的；而在storage模塊內(nèi)部，RDD又被視為由不同的block組成，對于RDD的存取是以block為單位進(jìn)行的，本質(zhì)上partition和block是等價的，只是看待的角度不同。在Spark storage模塊中中存取數(shù)據(jù)的最小單位是block，所有的操作都是以block為單位進(jìn)行的。

BlockManager對象被創(chuàng)建的時候會創(chuàng)建出MemoryStore和DiskStore對象用以存取block，如果內(nèi)存中擁有足夠的內(nèi)存，就使用 MemoryStore存儲，如果不夠，就 spill 到磁盤中，通過 DiskStore進(jìn)行存儲。

DiskStore 有一個DiskBlockManager,DiskBlockManager 主要用來創(chuàng)建并持有邏輯 blocks 與磁盤上的 blocks之間的映射，一個邏輯 block 通過 BlockId 映射到一個磁盤上的文件。在 DiskStore 中會調(diào)用 diskManager.getFile 方法，如果子文件夾不存在，會進(jìn)行創(chuàng)建，文件夾的命名方式為(spark-local-yyyyMMddHHmmss-xxxx, xxxx是一個隨機(jī)數(shù))，所有的block都會存儲在所創(chuàng)建的folder里面。
MemoryStore 相對于DiskStore需要根據(jù)block id hash計算出文件路徑并將block存放到對應(yīng)的文件里面，MemoryStore管理block就顯得非常簡單：MemoryStore內(nèi)部維護(hù)了一個hash map來管理所有的block，以block id為key將block存放到hash map中。而從MemoryStore中取得block則非常簡單，只需從hash map中取出block id對應(yīng)的value即可。

BlockManager 的 PUT 和GET接口

BlockManager 提供了 Put 接口和 Get 接口，這兩個 api 屏蔽了底層的細(xì)節(jié)，我們來看下底層是如何實(shí)現(xiàn)的

GET操作如果 local 中存在就直接返回，從本地獲取一個Block, 會先判斷如果是 useMemory，直接從內(nèi)存中取出，如果是 useDisk，會從磁盤中取出返回，然后根據(jù)useMemory判斷是否在內(nèi)存中緩存一下，方便下次獲取，如果local 不存在，從其他節(jié)點(diǎn)上獲取，當(dāng)然元信息是存在 drive上的，要根據(jù)我們上文中提到的 GETlocation 協(xié)議獲取 Block 所在節(jié)點(diǎn)位置，然后到其他節(jié)點(diǎn)上獲取。
PUT操作操作之前會加鎖來避免多線程的問題，存儲的時候會根據(jù) 存儲級別，調(diào)用對應(yīng)的是 memoryStore 還是 diskStore，然后在具體存儲器上面調(diào)用存儲接口。如果有 replication 需求，會把數(shù)據(jù)備份到其他的機(jī)器上面。

blockManager 和 blockTransferService 關(guān)系

spark 歷史上使用過兩套網(wǎng)絡(luò)框架，最開始的時候， rpc 調(diào)用使用的是 akka, 大文件傳輸使用的是 netty, 后面統(tǒng)一全部使用 netty, 這里的大文件傳輸其實(shí)走的是 netty, 在啟動 blockManager的時候會啟動一個 blockTransferService 服務(wù)，這個服務(wù)就是用來傳輸大文件用的，對應(yīng)的具體類是 NettyBlockTransferService，這個實(shí)例中也會有 BlocakManager的引用，會啟動一個 NettyBlockRpcServer的 netty Handler, 也擁有 BlocakManager 的引用，用來提供服務(wù)， BlocakManager 根據(jù) BlockId 獲取一個 Block 然后包裝為一個 ManagedBuffer 對象，

當(dāng)我們需要從遠(yuǎn)端獲取一個 Block的時候，就需要 blockTransferService 傳輸大的字節(jié)數(shù)組，

首先需要從 driver上獲取到 Block的真正存儲位置，然后調(diào)用 blockTransferService 的 fetchBlocks方法，去其他真正存儲節(jié)點(diǎn)上去fetch數(shù)據(jù)，會從 client 資源池中獲取一個client, 如果是一對一的進(jìn)行fetch，使用的是 OneForOneBlockFetcher，這個Fetcher 是以 Chunks 為單位分別單獨(dú)fetch, 每個 Chunks 也就對應(yīng)一個Block的數(shù)據(jù)，根據(jù)配置，會進(jìn)行重試直到***重試次數(shù)，發(fā)送 OpenBlocks消息，里面會包裝對應(yīng)的是哪個 BlockId, 其他節(jié)點(diǎn)服務(wù)端會根據(jù) BlockId 從 blockManager中拿到數(shù)據(jù)，然后用來傳輸，使用的是 netty 的流式傳輸方式，同時也會有回調(diào)函數(shù)，

如果是備份的時候同步上傳一個 Block, 其他節(jié)點(diǎn)服務(wù)端會根據(jù)，uploadBlock消息中包含的BlockId, 在本地的BlockManager 中冗余存儲一份，

ChunkFetch也有一個類似Stream的概念，ChunkFetch的對象是“一個內(nèi)存中的Iterator[ManagedBuffer]”，即一組Buffer，每一個Buffer對應(yīng)一個chunkIndex，整個Iterator[ManagedBuffer]由一個StreamID標(biāo)識。Client每次的ChunkFetch請求是由（streamId，chunkIndex）組成的唯一的StreamChunkId，Server端根據(jù)StreamChunkId獲取為一個Buffer并返回給Client；不管是Stream還是ChunkFetch，在Server的內(nèi)存中都需要管理一組由StreamID與資源之間映射，即StreamManager類，它提供了getChunk和openStream兩個接口來分別響應(yīng)ChunkFetch與Stream兩種操作，并且針對Server的ChunkFetch提供一個registerStream接口來注冊一組Buffer，比如可以將BlockManager中一組BlockID對應(yīng)的Iterator[ManagedBuffer]注冊到StreamManager，從而支持遠(yuǎn)程Block Fetch操作。

對于ExternalShuffleService(一種單獨(dú)shuffle服務(wù)進(jìn)程，對其他計算節(jié)點(diǎn)提供本節(jié)點(diǎn)上面的所有shuffle map輸出)，它為遠(yuǎn)程Executor提供了一種OpenBlocks的RPC接口，即根據(jù)請求的appid，executorid，blockid(appid+executor對應(yīng)本地一組目錄，blockid拆封出)從本地磁盤中加載一組FileSegmentManagedBuffer到內(nèi)存，并返回加載后的streamId返回給客戶端，從而支持后續(xù)的ChunkFetch的操作。

Partition 與 Block 的關(guān)系

我們都知道， RDD 的運(yùn)算是基于 partition，每個 task 代表一個分區(qū)上一個 stage 內(nèi)的運(yùn)算閉包， task 被分別調(diào)度到多個 executor上去運(yùn)行，那么是在哪里變成了 Block 呢，我們以 spark 2.11 源碼為準(zhǔn)，看看這個轉(zhuǎn)變過程，

一個 RDD 調(diào)度到 executor 上會運(yùn)行調(diào)用 getOrCompute方法，

  
 
 
 
   
  
  
  SparkEnv.get.blockManager.getOrElseUpdate(blockId, storageLevel, elementClassTag, () => {
   
  
  
        readCachedBlock = false
   
  
  
        computeOrReadCheckpoint(partition, context)
   
  
  
      })

如果 Block 在 BlockManager 中存在，就會從 BlockManager 中獲取，如果不存在，就進(jìn)行計算這個Block, 然后在 BlockManager 中進(jìn)行存儲持久化，方便下次使用，

當(dāng)然獲取的時候是先從本地的 BlockManager 中獲取，如果本地沒有，然后再從 remote 獲取，先從 driver 上獲取到元數(shù)據(jù) Block的位置，然后去到真正的節(jié)點(diǎn)上fetch

如果沒有，就進(jìn)行計算，然后根據(jù)存儲級別，存儲到計算節(jié)點(diǎn)本地的BlockManager 的內(nèi)存或磁盤中，

這樣RDD的transformation、action就和block數(shù)據(jù)建立了聯(lián)系，雖然抽象上我們的操作是在partition層面上進(jìn)行的，但是partition最終還是被映射成為block，因此實(shí)際上我們的所有操作都是對block的處理和存取。

blockManager 在 spark 中扮演的角色

blockManager 是非常非常重要的一個 spark 組件，我們隨便舉幾個例子，你就知道 BlockManager 多重要了，

spark shuffle 的過程總用到了 BlockManager 作為數(shù)據(jù)的中轉(zhuǎn)站
spark broadcast 調(diào)度 task 到多個 executor 的時候， broadCast 底層使用的數(shù)據(jù)存儲層
spark streaming 一個 ReceiverInputDStream 接受到的數(shù)據(jù)也是先放在 BlockManager 中，然后封裝為一個 BlockRdd 進(jìn)行下一步運(yùn)算的
如果我們對一個 rdd 進(jìn)行了cache, cacheManager 也是把數(shù)據(jù)放在了 blockmanager 中，截斷了計算鏈依賴，后續(xù)task 運(yùn)行的時候可以直接從 cacheManager 中獲取到 cacherdd ，不用再從頭計算。

spark cache 與 spark broadcast task

我隨便舉兩個例子，看看具體 spark cache 和 spark broadcast 調(diào)度 task 的時候怎么用的 blockManager的

spark cache

rdd 計算的時候，首先根據(jù)RDD id和partition index構(gòu)造出block id (rdd_xx_xx)，接著從BlockManager中取出相應(yīng)的block，如果該block存在，表示此RDD在之前已經(jīng)被計算過和存儲在BlockManager中，因此取出即可，無需再重新計算。如果 block 不存在我們可以計算出來，然后吧 block 通過 doPutIterator 函數(shù)存儲在節(jié)點(diǎn)上的 BlockManager上面，匯報block信息到 driver, 下次如果使用同一個 rdd, 就可以直接從分布式存儲中直接取出相應(yīng)的 block

下面看一下源碼

  
 
 
 
   
  
  
  final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
   
  
  
      if (storageLevel != StorageLevel.NONE) {
   
  
  
        getOrCompute(split, context)
   
  
  
      } else {
   
  
  
        computeOrReadCheckpoint(split, context)
   
  
  
      }
   
  
  
    }

如果存儲級別不是 NONE類型就會調(diào)用 getOrCompute 這個我們已經(jīng)看過了，里面實(shí)際調(diào)用 SparkEnv.get.blockManager.getOrElseUpdate 方法，如果 Block 在 BlockManager 中存在，就會從 BlockManager 中獲取，如果不存在，就進(jìn)行計算這個Block, 然后在 BlockManager 中進(jìn)行存儲持久化，方便下次使用，

在 BlockManager 進(jìn)行存儲后，會調(diào)用下面的代碼把匯報block信息到 driver，

  
 
 
 
   
  
  
  private def tryToReportBlockStatus(
   
  
  
       blockId: BlockId,
   
  
  
       status: BlockStatus,
   
  
  
       droppedMemorySize: Long = 0L): Boolean = {
   
  
  
     val storageLevel = status.storageLevel
   
  
  
     val inMemSize = Math.max(status.memSize, droppedMemorySize)
   
  
  
     val onDiskSize = status.diskSize
   
  
  
     master.updateBlockInfo(blockManagerId, blockId, storageLevel, inMemSize, onDiskSize)
   
  
  
   }

實(shí)際上上想 masterEndpoint 的引用發(fā)送一條 UpdateBlockInfo消息， master 會把這個 blockId 對應(yīng)的 location 放在 driver 上，

同樣的如果一個 Block已經(jīng)計算過了，會到 driver 上獲取到 location 信息

  
 
 
 
   
  
  
  private def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
   
  
  
     val locs = Random.shuffle(master.getLocations(blockId))
   
  
  
     val (preferredLocs, otherLocs) = locs.partition { loc => blockManagerId.host == loc.host }
   
  
  
     preferredLocs ++ otherLocs
   
  
  
   }

spark broadcast task

這個調(diào)度 task 到多個 task 上面過程代碼太多，我就不貼了，直接說一下流程，

DAGScheduler 在 submitMissingTasks 方法提交 task的時候，會把 task 包裝為一個 Broadcast 類型，里面使用 TorrentBroadcastFactory 創(chuàng)建一個 TorrentBroadcast 的類型，使用的是p2p的協(xié)議，會減輕 master 的壓力，這個里面會調(diào)用 writeBlocks 里面把taskBinary 通過 blockManager.putSingle 放在 BlockManager 緩存中
ShuffleMapTask 或者 ResultTask，然后調(diào)用 runTask 方法, 里面實(shí)際上會調(diào)用 Broadcast 的value 方法，里面最終調(diào)用了 BlockManager 的 getLocalBytes 或者 getRemoteBytes 方法

blockManager 在 spark streaming 中的應(yīng)用

ReceiverTracker 在啟動的時候，會運(yùn)行一個 job, 這個job 就是到各個executor上去啟動 ReceiverSupervisorImpl，然后啟動各個具體的數(shù)據(jù)接收器，如果是SocketInputDStream，就會啟動一個 SocketReceiver，
Receiver 接收到數(shù)據(jù)后，先在 BlockGenerator 中緩存，等到達(dá)一定的大小后，調(diào)用 BlockManagerBasedBlockHandler 的 storeBlock方法持久化到 BlockManager 中，然后把數(shù)據(jù)信息匯報到 ReceiverTracker上，最終匯總到 ReceivedBlockTracker 中的 timeToAllocatedBlocks中，
ReceiverInputDStream compute的時候， receivedBlockTracker 會根據(jù)時間獲取到 BlockManager 中的元信息，里面最終對應(yīng)的還是 BlockManager 的存儲位置，最終獲取到數(shù)據(jù)進(jìn)行計算，

測試 blockManager

我們做一個簡單的測試，兩端代碼的區(qū)別就是一個進(jìn)行了cache ，一個沒有進(jìn)行cache。

  
 
 
 
   
  
  
  val file = sc.textFile("/fusionlog/midsourcenew/2017-03-13-18-15_2.gz") 
   
  
  
  file.count() 
   
  
  
  file.count()

我們從日志可以觀察出來， ***段代碼，兩個 job 中都從 hdfs 中讀取文件，讀取了兩次，

  
 
 
 
   
  
  
  val file = sc.textFile("/fusionlog/midsourcenew/2017-03-13-18-15_2.gz").cache()
   
  
  
  file.count()
   
  
  
  file.count()

有以下日志

  
 
 
 
   
  
  
  MemoryStore: Block rdd_1_0 stored as values in memory (estimated size 1354.9 MB, free 4.9 GB)
   
  
  
  BlockManager: Found block rdd_1_0 locally

我們發(fā)現(xiàn)在***次讀取文件后，把文件 cache 在了 blockManager 中，下一個 job 運(yùn)行的時候，在本地 BlockManager 直接發(fā)現(xiàn)獲取到了 block ，沒有讀取 hdfs 文件，

在 spark ui 中也發(fā)現(xiàn)了 cache的 Block，全部是在內(nèi)存中緩存的，

本文題目：spark自己的分布式存儲系統(tǒng)-BlockManager
鏈接URL：http://www.fisionsoft.com.cn/article/djsdgoo.html

新聞中心

Spark 自己的分布式存儲系統(tǒng) - BlockManager

其他資訊