主要负责整个应用程序在运行期间block元数据的管理和维护，以及向从节点发送指令执行命令。

一核心属性

RpcEndpointRef: driverEndpointBlockManagerMasterEndpoint通信终端

Boolean isDriver: 是否在Driver

二重要方法

2.1 从driver通信终端删除一个executor

defremoveExecutor(execId:String) {
// 向BlockManagerMasterEndpoint发送RemoveExecutor消息
tell(RemoveExecutor(execId))
logInfo("Removed "+ execId + " successfully in removeExecutor")
}

2.2 向BlockManagerMasterEndpoint发送RegisterBlockManager消息

def registerBlockManager(blockManagerId: BlockManagerId, maxMemSize: Long, slaveEndpoint: RpcEndpointRef): Unit = {logInfo("Trying to register BlockManager")// 向BlockManagerMasterEndpoint发送RegisterBlockManager消息tell(RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint))logInfo("Registered BlockManager")
}

2.3 更新数据块信息

def updateBlockInfo(blockManagerId: BlockManagerId,blockId: BlockId,storageLevel: StorageLevel,memSize: Long,diskSize: Long,externalBlockStoreSize: Long): Boolean = {// 向BlockManagerMasterEndpoint发送UpdateBlockInfo消息，并且返回结果val res = driverEndpoint.askWithRetry[Boolean](UpdateBlockInfo(blockManagerId, blockId, storageLevel,memSize, diskSize, externalBlockStoreSize))logDebug(s"Updated info of block $blockId")res
}

2.4 获取某个block的所有BlockManagerId信息

def getLocations(blockId: BlockId): Seq[BlockManagerId] = {// 向BlockManagerMasterEndpoint发送GetLocations消息driverEndpoint.askWithRetry[Seq[BlockManagerId]](GetLocations(blockId))
}

2.5 获取多个block的位置信息

def getLocations(blockIds: Array[BlockId]): IndexedSeq[Seq[BlockManagerId]] = {// 向BlockManagerMasterEndpoint发送GetLocationsMultipleBlockIds消息driverEndpoint.askWithRetry[IndexedSeq[Seq[BlockManagerId]]](GetLocationsMultipleBlockIds(blockIds))
}

2.6 检查是否存在该数据块

def contains(blockId: BlockId): Boolean = {!getLocations(blockId).isEmpty
}

2.7 获取Executor的非当前BlockManagerId

def getPeers(blockManagerId: BlockManagerId): Seq[BlockManagerId] = {driverEndpoint.askWithRetry[Seq[BlockManagerId]](GetPeers(blockManagerId))
}

2.8 向BlockManagerMasterEndpoint发送RemoveBlock消息删除数据块

def removeBlock(blockId: BlockId) {driverEndpoint.askWithRetry[Boolean](RemoveBlock(blockId))
}

2.9 删除指定RDD的所有数据块

def removeRdd(rddId: Int, blocking: Boolean) {val future = driverEndpoint.askWithRetry[Future[Seq[Int]]](RemoveRdd(rddId))future.onFailure {case e: Exception =>logWarning(s"Failed to remove RDD $rddId - ${e.getMessage}", e)}(ThreadUtils.sameThread)if (blocking) {timeout.awaitResult(future)}
}

2.10 删除指定shuffle的所有数据块

def removeShuffle(shuffleId: Int, blocking: Boolean) {val future = driverEndpoint.askWithRetry[Future[Seq[Boolean]]](RemoveShuffle(shuffleId))future.onFailure {case e: Exception =>logWarning(s"Failed to remove shuffle $shuffleId - ${e.getMessage}", e)}(ThreadUtils.sameThread)if (blocking) {timeout.awaitResult(future)}
}

2.11 返回每一个block manager的内存状态
def getMemoryStatus: Map[BlockManagerId, (Long, Long)] = {driverEndpoint.askWithRetry[Map[BlockManagerId, (Long, Long)]](GetMemoryStatus)
}

2.12 返回存储状态

def getStorageStatus: Array[StorageStatus] = {driverEndpoint.askWithRetry[Array[StorageStatus]](GetStorageStatus)
}

2.13 获取block状态

def getBlockStatus(blockId: BlockId,askSlaves: Boolean = true): Map[BlockManagerId, BlockStatus] = {val msg = GetBlockStatus(blockId, askSlaves)val response = driverEndpoint.askWithRetry[Map[BlockManagerId, Future[Option[BlockStatus]]]](msg)val (blockManagerIds, futures) = response.unzipimplicit val sameThread = ThreadUtils.sameThreadval cbf =implicitly[CanBuildFrom[Iterable[Future[Option[BlockStatus]]],Option[BlockStatus],Iterable[Option[BlockStatus]]]]val blockStatus = timeout.awaitResult(Future.sequence[Option[BlockStatus], Iterable](futures)(cbf, ThreadUtils.sameThread))if (blockStatus == null) {throw new SparkException("BlockManager returned null for BlockStatus query: " + blockId)}blockManagerIds.zip(blockStatus).flatMap { case (blockManagerId, status) =>status.map { s => (blockManagerId, s) }}.toMap
}

2.14 找到那些executor有缓存的block

def hasCachedBlocks(executorId: String): Boolean = {driverEndpoint.askWithRetry[Boolean](HasCachedBlocks(executorId))
}

Spark源码分析之BlockManagerMaster相关推荐

Spark 源码分析
2019独角兽企业重金招聘Python工程师标准>>> 一. 启动篇 (一) 引子在spark-shell终端执行 val arr = Array(1,2,3,4) val rdd ...
spark 源码分析 Blockmanager
原文链接参考, Spark源码分析之-Storage模块对于storage, 为何Spark需要storage模块?为了cache RDD Spark的特点就是可以将RDD cache在memo ...
Spark源码分析 – DAGScheduler
DAGScheduler的架构其实非常简单, 1. eventQueue, 所有需要DAGScheduler处理的事情都需要往eventQueue中发送event 2. eventLoop Threa ...
spark 源码分析之十八 -- Spark存储体系剖析
本篇文章主要剖析BlockManager相关的类以及总结Spark底层存储体系. 总述先看 BlockManager相关类之间的关系如下: 我们从NettyRpcEnv 开始,做一下简单说明. Ne ...
Spark源码分析之七：Task运行（一）
在Task调度相关的两篇文章<Spark源码分析之五:Task调度(一)>与<Spark源码分析之六:Task调度(二)>中,我们大致了解了Task调度相关的主要逻辑,并且在T ...
Spark源码分析之九：内存管理模型
Spark是现在很流行的一个基于内存的分布式计算框架,既然是基于内存,那么自然而然的,内存的管理就是Spark存储管理的重中之重了.那么,Spark究竟采用什么样的内存管理模型呢?本文就为大家揭开Sp ...
spark 源码分析之二十 -- Stage的提交
引言上篇 spark 源码分析之十九 -- DAG的生成和Stage的划分中,主要介绍了下图中的前两个阶段DAG的构建和Stage的划分. 本篇文章主要剖析,Stage是如何提交的. rdd的依赖 ...
spark 源码分析之十九 -- DAG的生成和Stage的划分
上篇文章 spark 源码分析之十八 -- Spark存储体系剖析重点剖析了 Spark的存储体系.从本篇文章开始,剖析Spark作业的调度和计算体系. 在说DAG之前,先简单说一下RDD. 对RD ...
Spark源码分析：多种部署方式之间的区别与联系
作者:过往记忆从官方的文档我们可以知道, Spark 的部署方式有很多种:local.Standalone.Mesos.YARN-..不同部署方式的后台处理进程是不一样的,但是如果我们从代码的角度来 ...

Spark源码分析之BlockManagerMaster

一核心属性

二重要方法

2.1 从driver通信终端删除一个executor

2.2 向BlockManagerMasterEndpoint发送RegisterBlockManager消息

2.3 更新数据块信息

2.4 获取某个block的所有BlockManagerId信息

2.5 获取多个block的位置信息

2.6 检查是否存在该数据块

2.7 获取Executor的非当前BlockManagerId

2.8 向BlockManagerMasterEndpoint发送RemoveBlock消息删除数据块

2.9 删除指定RDD的所有数据块

2.10 删除指定shuffle的所有数据块

2.12 返回存储状态

2.13 获取block状态

2.14 找到那些executor有缓存的block

Spark源码分析之BlockManagerMaster相关推荐

最新文章

热门文章

Spark源码分析之BlockManagerMaster

一 核心属性

二 重要方法

2.1 从driver通信终端删除一个executor

2.2 向BlockManagerMasterEndpoint发送RegisterBlockManager消息

2.3 更新数据块信息

2.4 获取某个block的所有BlockManagerId信息

2.5 获取多个block的位置信息

2.6 检查是否存在该数据块

2.7 获取Executor的非当前BlockManagerId

2.8 向BlockManagerMasterEndpoint发送RemoveBlock消息删除数据块

2.9 删除指定RDD的所有数据块

2.10 删除指定shuffle的所有数据块

2.12 返回存储状态

2.13 获取block状态

2.14 找到那些executor有缓存的block

Spark源码分析之BlockManagerMaster相关推荐

最新文章

热门文章

一核心属性

二重要方法