spark源码学习（十）--- blockManager分析

blockManager主要原理：

blockmanager位于org.apache.spark.storage中，包含四个重要的组件：DiskStore，MemoryStore，Blocktransferservice，ConnectionManager。其中，diskStore负责对磁盘上的数据读写；memoryStore负责内存数据的读写，connectionManager负责到远程节点的连接，BlockManagerWorker负责读写远程节点的的数据。当blockManager启动创建后会向blockManagerMaster注册，其中blockManagerMaster位于driver上，管理者数据的元数据，比如包含了blockmanagerInfo，blockStatus，当blockManagerMaster进行了增删改操作，blockManager会通知blockManagerMaster，blockManagerMaster通过blockManagerInfo内的blockStatus进行元数据的操作。

首先看位于org.apache.spark.storage中的blockManagerMaster，重要的功能在BlockManagerMasterActor类中定义，下面分析blockManagerMasterInfo类：

首先，持有一个blockManagerInfo的hashmap，记录了BlockManagerId与BlockManagerInfo的映射，BlockManagerInfo记录blockManager的一些元数据信息：

private val blockManagerInfo = new mutable.HashMap[BlockManagerId, BlockManagerInfo]

另外一个重要的成员映射，executor与blockManager的映射：

private val blockManagerIdByExecutor = new mutable.HashMap[String, BlockManagerId]

下面来看blockManager的注册：

  private def register(id: BlockManagerId, maxMemSize: Long, slaveActor: ActorRef) {val time = System.currentTimeMillis()//如果没有注册过，则去注册blockManagerif (!blockManagerInfo.contains(id)) {// BlockManagerId包含有成员变量executorID，通过BlockManagerId找到executorID// 然后判断该executorID是否存在，如果存在，那么将存在的该executorid对应的BlockManagerId移除// 因为此处是在!blockManagerInfo.contains(id)这个条件下，所以必须没有该executorid对应的BlockManagerIdblockManagerIdByExecutor.get(id.executorId) match {case Some(oldId) =>// A block manager of the same executor already exists, so remove it (assumed dead)logError("Got two different block manager registrations on same executor - " + s" will replace old one $oldId with new one $id")removeExecutor(id.executorId)  case None =>}logInfo("Registering block manager %s with %s RAM, %s".format(id.hostPort, Utils.bytesToString(maxMemSize), id))//将新的executorID与BlockManagerId映射起来，key为executorId，value为BlockManagerIdblockManagerIdByExecutor(id.executorId) = id//生成blockManagerInfo与BlockManagerId的映射blockManagerInfo(id) = new BlockManagerInfo(id, System.currentTimeMillis(), maxMemSize, slaveActor)}listenerBus.post(SparkListenerBlockManagerAdded(time, id, maxMemSize))}

更新blockInfo，每个blockmanager上，如果block发生了变化都会调用updateBlockInfo进行blockInfo的更新：

  private def updateBlockInfo(blockManagerId: BlockManagerId,blockId: BlockId,storageLevel: StorageLevel,memSize: Long,diskSize: Long,tachyonSize: Long): Boolean = {if (!blockManagerInfo.contains(blockManagerId)) {if (blockManagerId.isDriver && !isLocal) {// We intentionally do not register the master (except in local mode),// so we should not indicate failure.return true} else {return false}}if (blockId == null) {blockManagerInfo(blockManagerId).updateLastSeenMs()return true}blockManagerInfo(blockManagerId).updateBlockInfo(blockId, storageLevel, memSize, diskSize, tachyonSize)var locations: mutable.HashSet[BlockManagerId] = nullif (blockLocations.containsKey(blockId)) {locations = blockLocations.get(blockId)} else {locations = new mutable.HashSet[BlockManagerId]blockLocations.put(blockId, locations)}if (storageLevel.isValid) {locations.add(blockManagerId)} else {locations.remove(blockManagerId)}// Remove the block from master tracking if it has been removed on all slaves.if (locations.size == 0) {blockLocations.remove(blockId)}true}

下面看blockManager类，首先，来看blockManager的类定义：

private[spark] class BlockManager(executorId: String,actorSystem: ActorSystem,val master: BlockManagerMaster,defaultSerializer: Serializer,maxMemory: Long,val conf: SparkConf,mapOutputTracker: MapOutputTracker,shuffleManager: ShuffleManager,blockTransferService: BlockTransferService,securityManager: SecurityManager,numUsableCores: Int)extends BlockDataManager with Logging

blockManager中管理的几种存储级别：内存，磁盘，tachyon，每种存储级别会有对应的类进行数据的操作，分别是memoryStore，diskStore，tachyonStore。

  private[spark] val memoryStore = new MemoryStore(this, maxMemory)private[spark] val diskStore = new DiskStore(this, diskBlockManager)private[spark] lazy val tachyonStore: TachyonStore = {val storeDir = conf.get("spark.tachyonStore.baseDir", "/tmp_spark_tachyon")val appFolderName = conf.get("spark.tachyonStore.folderName")val tachyonStorePath = s"$storeDir/$appFolderName/${this.executorId}"val tachyonMaster = conf.get("spark.tachyonStore.url",  "tachyon://localhost:19998")val tachyonBlockManager =new TachyonBlockManager(this, tachyonStorePath, tachyonMaster)tachyonInitialized = truenew TachyonStore(this, tachyonBlockManager)}

在blockManager初始化的时候回调用initialize方法：

  def initialize(appId: String): Unit = {blockTransferService.init(this)shuffleClient.init(appId)//一个blockManager对应一个executorId，blockTransferService的host，portblockManagerId = BlockManagerId(executorId, blockTransferService.hostName, blockTransferService.port)shuffleServerId = if (externalShuffleServiceEnabled) {BlockManagerId(executorId, blockTransferService.hostName, externalShuffleServicePort)} else {blockManagerId}//像BlockManagerMaster注册blockManagermaster.registerBlockManager(blockManagerId, maxMemory, slaveActor)// Register Executors' configuration with the local shuffle service, if one should exist.if (externalShuffleServiceEnabled && !blockManagerId.isDriver) {registerWithExternalShuffleServer()}}

blockManager获取数据的方法doGetLocal：

首先来看读取内存存储数据的情况：

  private def doGetLocal(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {//orNull：option方法，如果它不为空返回该选项的值，如果它是空则返回null。//blockInfo：TimeStampedHashMap[BlockId, BlockInfo]val info = blockInfo.get(blockId).orNullif (info != null) {info.synchronized {// Double check to make sure the block is still there. There is a small chance that the// block has been removed by removeBlock (which also synchronizes on the blockInfo object).// Note that this only checks metadata tracking. If user intentionally deleted the block// on disk or from off heap storage without using removeBlock, this conditional check will// still pass but eventually we will get an exception because we can't find the block.//判断blockInfo是否为空，blockInfo记录了block的元数据信息//如果通过调用程序来移除block，比如认为操作移除block的话，会发生此处的情况if (blockInfo.get(blockId).isEmpty) {logWarning(s"Block $blockId had been removed")return None}// If another thread is writing the block, wait for it to become ready.//如果其他线程正在操作该block ，那么等待if (!info.waitForReady()) {// If we get here, the block write failed.logWarning(s"Block $blockId was marked as failure.")return None}//获取存储级别，内存、tachyon、是否内存或者tachyon沾满后会刷到磁盘，是否需要多个副本val level = info.levellogDebug(s"Level for block $blockId is $level")// Look for the block in memory//数据存储在内存的情况//调用memoryStore的getValues与getBytes来读取数据if (level.useMemory) {logDebug(s"Getting block $blockId from memory")val result = if (asBlockResult) {//需要的是非序列化的数据memoryStore.getValues(blockId).map(new BlockResult(_, DataReadMethod.Memory, info.size))} else {//需要的是序列化的数据memoryStore.getBytes(blockId)}result match {case Some(values) =>return resultcase None =>logDebug(s"Block $blockId not found in memory")}}

这里根据获取的数据是否需要序列化来分别调用getValues和getBytes方法，getValues获取的是非序列化数据：

  override def getValues(blockId: BlockId): Option[Iterator[Any]] = {val entry = entries.synchronized {entries.get(blockId)}if (entry == null) {None} else if (entry.deserialized) {//非序列化数据。直接返回Some(entry.value.asInstanceOf[Array[Any]].iterator)} else {//序列化数据，反序列化后返回val buffer = entry.value.asInstanceOf[ByteBuffer].duplicate() // Doesn't actually copy dataSome(blockManager.dataDeserialize(blockId, buffer))}}

getBytes获取的是序列化数据：

  override def getBytes(blockId: BlockId): Option[ByteBuffer] = {val entry = entries.synchronized {//从内存中获取数据entries.get(blockId)}if (entry == null) {None} else if (entry.deserialized) {// 如果获取的数据是非序列化的数据，那么序列化数据后返回，否则直接返回Some(blockManager.dataSerialize(blockId, entry.value.asInstanceOf[Array[Any]].iterator))} else {Some(entry.value.asInstanceOf[ByteBuffer].duplicate()) // Doesn't actually copy the data}}

下面分析从磁盘读取数据的情况，分为两种：一是只使用磁盘，二是数据既使用了磁盘也使用了内存：

        if (level.useDisk) {logDebug(s"Getting block $blockId from disk")val bytes: ByteBuffer = diskStore.getBytes(blockId) match {case Some(b) => bcase None =>throw new BlockException(blockId, s"Block $blockId not found on disk, though it should be")}assert(0 == bytes.position())//如果只使用磁盘没有使用内存if (!level.useMemory) {// If the block shouldn't be stored in memory, we can just return itif (asBlockResult) {return Some(new BlockResult(dataDeserialize(blockId, bytes), DataReadMethod.Disk,info.size))} else {return Some(bytes)}//如果使用磁盘和内存混合存储} else {// Otherwise, we also have to store something in the memory storeif (!level.deserialized || !asBlockResult) {/* We'll store the bytes in memory if the block's storage level includes* "memory serialized", or if it should be cached as objects in memory* but we only requested its serialized bytes. */val copyForMemory = ByteBuffer.allocate(bytes.limit)copyForMemory.put(bytes)memoryStore.putBytes(blockId, copyForMemory, level)bytes.rewind()}if (!asBlockResult) {return Some(bytes)} else {val values = dataDeserialize(blockId, bytes)if (level.deserialized) {// Cache the values before returning themval putResult = memoryStore.putIterator(blockId, values, level, returnValues = true, allowPersistToDisk = false)// The put may or may not have succeeded, depending on whether there was enough// space to unroll the block. Either way, the put here should return an iterator.putResult.data match {case Left(it) =>return Some(new BlockResult(it, DataReadMethod.Disk, info.size))case _ =>// This only happens if we dropped the values back to disk (which is never)throw new SparkException("Memory store did not return an iterator!")}} else {return Some(new BlockResult(values, DataReadMethod.Disk, info.size))}}}}}} else {logDebug(s"Block $blockId not registered locally")}None}

上面是从本地读取数据的情况源码分析，除此之外还有从远程读取数据的情况，远程读取数据的情况在doGetRomote中：

  private def doGetRemote(blockId: BlockId, asBlockResult: Boolean): Option[Any] = {//判断，如果条件不满足，则抛出异常require(blockId != null, "BlockId is null")//打乱block所在位置，以便均衡val locations = Random.shuffle(master.getLocations(blockId))//循环读取所有位置的数据for (loc <- locations) {logDebug(s"Getting remote block $blockId from $loc")//远程读取数据val data = blockTransferService.fetchBlockSync(loc.host, loc.port, loc.executorId, blockId.toString).nioByteBuffer()if (data != null) {if (asBlockResult) {//返回的是序列化的数据，如果不需要序列化，则进行反序列化return Some(new BlockResult(dataDeserialize(blockId, data),DataReadMethod.Network,data.limit()))} else {return Some(data)}}logDebug(s"The value of block $blockId is null")}logDebug(s"Block $blockId not found")None}

以上分析的书读数据的两种情况：读取本地数据和读取远程数据。下面分析写数据，写数据由doPut方法来管理：

  private def doPut(blockId: BlockId,data: BlockValues,level: StorageLevel,tellMaster: Boolean = true,effectiveStorageLevel: Option[StorageLevel] = None): Seq[(BlockId, BlockStatus)] = {require(blockId != null, "BlockId is null")require(level != null && level.isValid, "StorageLevel is null or invalid")effectiveStorageLevel.foreach { level =>require(level != null && level.isValid, "Effective StorageLevel is null or invalid")}// Return value//blockStatus中封装了block的一些信息：/**        storageLevel: StorageLevel,memSize: Long,diskSize: Long,tachyonSize: Long*/val updatedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]/* Remember the block's storage level so that we can correctly drop it to disk if it needs* to be dropped right after it got put into memory. Note, however, that other threads will* not be able to get() this block until we call markReady on its BlockInfo. *///为将写入的block生成blockInfo并写入map中val putBlockInfo = {val tinfo = new BlockInfo(level, tellMaster)// Do atomically !//如果不存在该info信息，那么将blockId与 BlockInfo关联起来，放入mapval oldBlockOpt = blockInfo.putIfAbsent(blockId, tinfo)if (oldBlockOpt.isDefined) {if (oldBlockOpt.get.waitForReady()) {logWarning(s"Block $blockId already exists on this machine; not re-adding it")return updatedBlocks}// TODO: So the block info exists - but previous attempt to load it (?) failed.// What do we do now ? Retry on it ?oldBlockOpt.get} else {tinfo}}val startTimeMs = System.currentTimeMillis/* If we're storing values and we need to replicate the data, we'll want access to the values,* but because our put will read the whole iterator, there will be no values left. For the* case where the put serializes data, we'll remember the bytes, above; but for the case where* it doesn't, such as deserialized storage, let's rely on the put returning an Iterator. */var valuesAfterPut: Iterator[Any] = null// Ditto for the bytes after the putvar bytesAfterPut: ByteBuffer = null// Size of the block in bytesvar size = 0L// The level we actually use to put the blockval putLevel = effectiveStorageLevel.getOrElse(level)// If we're storing bytes, then initiate the replication before storing them locally.// This is faster as data is already serialized and ready to send.val replicationFuture = data match {case b: ByteBufferValues if putLevel.replication > 1 =>// Duplicate doesn't copy the bytes, but just creates a wrapperval bufferView = b.buffer.duplicate()Future { replicate(blockId, bufferView, putLevel) }case _ => null}//对blockInfo 加锁，多线程同步putBlockInfo.synchronized {logTrace("Put for block %s took %s to get into synchronized block".format(blockId, Utils.getUsedTimeMs(startTimeMs)))var marked = falsetry {// returnValues - Whether to return the values put// blockStore - The type of storage to put these values into// blockStore - 存储方式：内存磁盘还是tachyonval (returnValues, blockStore: BlockStore) = {//使用内存if (putLevel.useMemory) {// Put it in memory first, even if it also has useDisk set to true;// We will drop it to disk later if the memory store can't hold it.(true, memoryStore)//使用tachyon} else if (putLevel.useOffHeap) {// Use tachyon for off-heap storage(false, tachyonStore)//使用磁盘} else if (putLevel.useDisk) {// Don't get back the bytes from put unless we replicate them(putLevel.replication > 1, diskStore)} else {//否则，抛出没有指定正确的存储级别错误assert(putLevel == StorageLevel.NONE)throw new BlockException(blockId, s"Attempted to put block $blockId without specifying storage level!")}}// Actually put the values// 根据选择的store和数据类型，放入store中，putIterator方法写入数据并返回写入数据量等信息val result = data match {case IteratorValues(iterator) =>blockStore.putIterator(blockId, iterator, putLevel, returnValues)case ArrayValues(array) =>blockStore.putArray(blockId, array, putLevel, returnValues)case ByteBufferValues(bytes) =>bytes.rewind()blockStore.putBytes(blockId, bytes, putLevel)}size = result.sizeresult.data match {case Left (newIterator) if putLevel.useMemory => valuesAfterPut = newIteratorcase Right (newBytes) => bytesAfterPut = newBytescase _ =>}// Keep track of which blocks are dropped from memoryif (putLevel.useMemory) {result.droppedBlocks.foreach { updatedBlocks += _ }}//获取block对应的statusval putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)if (putBlockStatus.storageLevel != StorageLevel.NONE) {// Now that the block is in either the memory, tachyon, or disk store,// let other threads read it, and tell the master about it.marked = trueputBlockInfo.markReady(size)if (tellMaster) {//向master通知blockstatus，更新元数据信息reportBlockStatus(blockId, putBlockInfo, putBlockStatus)}updatedBlocks += ((blockId, putBlockStatus))}} finally {// If we failed in putting the block to memory/disk, notify other possible readers// that it has failed, and then remove it from the block info map.if (!marked) {// Note that the remove must happen before markFailure otherwise another thread// could've inserted a new BlockInfo before we remove it.blockInfo.remove(blockId)putBlockInfo.markFailure()logWarning(s"Putting block $blockId failed")}}}logDebug("Put block %s locally took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))// Either we're storing bytes and we asynchronously started replication, or we're storing// values and need to serialize and replicate them now:if (putLevel.replication > 1) {//数据副本数据大于1，那么复制多份数据data match {case ByteBufferValues(bytes) =>if (replicationFuture != null) {Await.ready(replicationFuture, Duration.Inf)}case _ =>val remoteStartTime = System.currentTimeMillis// Serialize the block if not already doneif (bytesAfterPut == null) {if (valuesAfterPut == null) {throw new SparkException("Underlying put returned neither an Iterator nor bytes! This shouldn't happen.")}bytesAfterPut = dataSerialize(blockId, valuesAfterPut)}replicate(blockId, bytesAfterPut, putLevel)//调用该方法复制数据logDebug("Put block %s remotely took %s".format(blockId, Utils.getUsedTimeMs(remoteStartTime)))}}BlockManager.dispose(bytesAfterPut)
<span style="white-space:pre"> </span>if (putLevel.replication > 1) {logDebug("Putting block %s with replication took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))} else {logDebug("Putting block %s without replication took %s".format(blockId, Utils.getUsedTimeMs(startTimeMs)))}updatedBlocks}

其中，实际写数据是由

 val result = data match {case IteratorValues(iterator) =>blockStore.putIterator(blockId, iterator, putLevel, returnValues)case ArrayValues(array) =>blockStore.putArray(blockId, array, putLevel, returnValues)case ByteBufferValues(bytes) =>bytes.rewind()blockStore.putBytes(blockId, bytes, putLevel)}

这段代码完成，blockStore根据存储级别分为三种：如果是memoryStore，写入的时候调用了memoryStore的putIterator方法，最后直到调用tryToPut方法：

  private def tryToPut(blockId: BlockId,value: Any,size: Long,deserialized: Boolean): ResultWithDroppedBlocks = {/* TODO: Its possible to optimize the locking by locking entries only when selecting blocks* to be dropped. Once the to-be-dropped blocks have been selected, and lock on entries has* been released, it must be ensured that those to-be-dropped blocks are not double counted* for freeing up more space for another block that needs to be put. Only then the actually* dropping of blocks (and writing to disk if necessary) can proceed in parallel. */var putSuccess = falseval droppedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]//并发同步，判断内存大小accountingLock.synchronized {//保证有可用的空间，该方法判断当前内存不足以存储当前数据，//那么同步entries那么移除一部分可以写到磁盘的数据，那么移除数据到磁盘//但是如果被移除的数据没有指定可以写到磁盘，那么此数据就丢了//移除的过程中，由于entries是一个linkedHashMap，所以移除的顺序是有限移除旧的entryval freeSpaceResult = ensureFreeSpace(blockId, size)val enoughFreeSpace = freeSpaceResult.successdroppedBlocks ++= freeSpaceResult.droppedBlocks//首先调用enoughFreeSpace方法判断内存是否够用if (enoughFreeSpace) {//实际放入的数据封装在MemoryEntry中val entry = new MemoryEntry(value, size, deserialized)entries.synchronized {//将新的数据entry放入到entries中，并将blockID与该entry对应entries.put(blockId, entry)currentMemory += size}val valuesOrBytes = if (deserialized) "values" else "bytes"logInfo("Block %s stored as %s in memory (estimated size %s, free %s)".format(blockId, valuesOrBytes, Utils.bytesToString(size), Utils.bytesToString(freeMemory)))putSuccess = true} else {//如果删除其他的数据还是不能放入数据，那么写入磁盘// Tell the block manager that we couldn't put it in memory so that it can drop it to// disk if the block allows disk storage.val data = if (deserialized) {Left(value.asInstanceOf[Array[Any]])} else {Right(value.asInstanceOf[ByteBuffer].duplicate())}val droppedBlockStatus = blockManager.dropFromMemory(blockId, data)droppedBlockStatus.foreach { status => droppedBlocks += ((blockId, status)) }}}ResultWithDroppedBlocks(putSuccess, droppedBlocks)}

如果是diskStore，则直接使用javaIO流写入磁盘。

数据的多副本操作定义如下：

    while (!done) {getRandomPeer() match {case Some(peer) =>try {val onePeerStartTime = System.currentTimeMillisdata.rewind()logTrace(s"Trying to replicate $blockId of ${data.limit()} bytes to $peer")//将数据异步写入其他的blockmanager上blockTransferService.uploadBlockSync(peer.host, peer.port, peer.executorId, blockId, new NioManagedBuffer(data), tLevel)logTrace(s"Replicated $blockId of ${data.limit()} bytes to $peer in %s ms".format(System.currentTimeMillis - onePeerStartTime))peersReplicatedTo += peerpeersForReplication -= peerreplicationFailed = falseif (peersReplicatedTo.size == numPeersToReplicateTo) {done = true  // specified number of peers have been replicated to}} catch {case e: Exception =>logWarning(s"Failed to replicate $blockId to $peer, failure #$failures", e)failures += 1replicationFailed = truepeersFailedToReplicateTo += peerif (failures > maxReplicationFailures) { // too many failures in replcating to peersdone = true}}case None => // no peer left to replicate todone = true}}

spark源码学习（十）--- blockManager分析相关推荐

Spark源码学习之IDEA源码阅读环境搭建
软件准备 (1)Java 1.8 (2)Scala 2.11.12(需要在IDEA中安装) (3)Maven 3.8.2(需要在IDEA中配置) (4)Git 2.33 以上软件需要安装好,并进行环境 ...
Box2d源码学习十四TOI之碰撞时间的实现
本系列博客是由扭曲45原创,欢迎转载,转载时注明出处,http://blog.csdn.net/cg0206/article/details/8441644 TOI全称Time of Impact,中 ...
Box2d源码学习十二b2Collision之碰撞(上)公共部分的实现
本系列博客是由扭曲45原创,欢迎转载,转载时注明出处,http://blog.csdn.net/cg0206/article/details/8390560 Box2d中将碰撞部分单独放到几个文件中去 ...
Spark源码性能优化案例分析
本篇文章枚举了几例常见的问题并给出了优化方案,推荐了两套测试性能优化工具问题: Spark 任务文件初始化调优资源分析,发现第一个 stage 时间特别长,耗时长达 14s , CPU 和网络通信 ...
spark 源码学习
转载: http://xuxping.com/2017/08/21/Spark%E6%9C%AC%E5%9C%B0%E5%BC%80%E5%8F%91%E7%8E%AF%E5%A2%83%E6%90% ...
libevent源码学习-----事件驱动流程分析
libevent中事件驱动的大体流程如下 /* 创建事件驱动 */ struct event_base* base = event_base_new(); /**创建一个事件*@param base: ...
spark 源码分析之十八 -- Spark存储体系剖析
本篇文章主要剖析BlockManager相关的类以及总结Spark底层存储体系. 总述先看 BlockManager相关类之间的关系如下: 我们从NettyRpcEnv 开始,做一下简单说明. Ne ...
spark 源码分析之二十 -- Stage的提交
引言上篇 spark 源码分析之十九 -- DAG的生成和Stage的划分中,主要介绍了下图中的前两个阶段DAG的构建和Stage的划分. 本篇文章主要剖析,Stage是如何提交的. rdd的依赖 ...
spark 源码分析 Blockmanager
原文链接参考, Spark源码分析之-Storage模块对于storage, 为何Spark需要storage模块?为了cache RDD Spark的特点就是可以将RDD cache在memo ...

spark源码学习（十）--- blockManager分析

spark源码学习（十）--- blockManager分析相关推荐

最新文章

热门文章