概述
BlockManager是spark自己的存储系统，RDD-Cache、 Shuffle-output、broadcast 等的实现都是基于BlockManager来实现的，BlockManager也是分布式结构，在driver和所有executor上都会有blockmanager节点，每个节点上存储的block信息都会汇报给driver端的blockManagerMaster作统一管理，BlockManager对外提供get和set数据接口，可将数据存储在memory, disk, off-heap。
blockManager的创建与注册
blockManagerMaster和blockManager都是在构造SparkEnv的时候创建的，Driver端是创建SparkContext的时候创建SparkEnv，Executor端的SparkEnv是在其守护进程CoarseGrainedExecutorBackend创建的时候创建的，下面看blockManager是怎么在sparkEnv中创建的：
// get&put 远程block的时候就是通过blockTransferService 完成的
val blockTransferService =
new NettyBlockTransferService(conf, securityManager, bindAddress, advertiseAddress,
blockManagerPort, numUsableCores)
val blockManagerMaster = new BlockManagerMaster(registerOrLookupEndpoint(
BlockManagerMaster.DRIVER_ENDPOINT_NAME,
new BlockManagerMasterEndpoint(rpcEnv, isLocal, conf, listenerBus)),
conf, isDriver)
// NB: blockManager is not valid until initialize() is called later.
val blockManager = new BlockManager(executorId, rpcEnv, blockManagerMaster,
serializerManager, conf, memoryManager, mapOutputTracker, shuffleManager,
blockTransferService, securityManager, numUsableCores)
构造blockManagerMaster的时候在Driver端是创建了一个BlockManagerMasterEndpoint并注册到了rpcEnv中，而在executor端是获取到了 Driver端BlockManagerMasterEndpoint的引用 BlockManagerMasterRef，以便后面的通信。随后都创建了自己blockManager，创建blockManager的时候都创建了BlockManagerSlaveEndpoint。
blockManager创建后还不能直接使用，接着都会调用blockManager的initialize方法，通过与master通信向master进行注册，master收到消息后会将blockManager的信息存到blockManagerInfo的map中，key为blockManagerId（保存着executorId、host、post等信息），value为BlockManagerInfo（保存着具体的block状态信息及 BlockManagerSlaveEndpoint 的引用 ），注册完后就可以真正干活了。
master与slave间的消息传递
slave -> master
// slave向master注册，会保存在master的blockManagerInfo中
case RegisterBlockManager(blockManagerId, maxMemSize, slaveEndpoint) =>
context.reply(register(blockManagerId, maxMemSize, slaveEndpoint))
// 一个Block的更新消息，BlockId作为一个Block的唯一标识，会保存Block所在的节点和位置关系，以及block 存储级别，大小 占用内存和磁盘大小
case _updateBlockInfo @
UpdateBlockInfo(blockManagerId, blockId, storageLevel, deserializedSize, size) =>
context.reply(updateBlockInfo(blockManagerId, blockId, storageLevel, deserializedSize, size))
listenerBus.post(SparkListenerBlockUpdated(BlockUpdatedInfo(_updateBlockInfo)))
// 用于获取指定 blockId 的 block 所在的 BlockManagerId 列表
case GetLocations(blockId) =>
context.reply(getLocations(blockId))
// 获取多个Block所在 的位置，位置中会反映Block位于哪个 executor, host 和端口
case GetLocationsMultipleBlockIds(blockIds) =>
context.reply(getLocationsMultipleBlockIds(blockIds))
// 一个block有可能在多个节点上存在，返回一个节点列表
case GetPeers(blockManagerId) =>
context.reply(getPeers(blockManagerId))
// 根据BlockId,获取所在executorEndpointRef 也就是 BlockManagerSlaveEndpoint的引用
case GetExecutorEndpointRef(executorId) =>
context.reply(getExecutorEndpointRef(executorId))
// 获取所有节点上的BlockManager的最大内存和剩余内存
case GetMemoryStatus =>
context.reply(memoryStatus)
// 获取所有节点上的BlockManager的最大磁盘空间和剩余磁盘空间
case GetStorageStatus =>
context.reply(storageStatus)
// 获取一个Block的状态信息，位置，占用内存和磁盘大小
case GetBlockStatus(blockId, askSlaves) =>
context.reply(blockStatus(blockId, askSlaves))
// 获取一个Block的存储级别和所占内存和磁盘大小
case GetMatchingBlockIds(filter, askSlaves) =>
context.reply(getMatchingBlockIds(filter, askSlaves))
// 删除Rdd对应的Block数据
case RemoveRdd(rddId) =>
context.reply(removeRdd(rddId))
// 删除 shuffleId对应的BlockId的Block
case RemoveShuffle(shuffleId) =>
context.reply(removeShuffle(shuffleId))
// 删除Broadcast对应的Block数据
case RemoveBroadcast(broadcastId, removeFromDriver) =>
context.reply(removeBroadcast(broadcastId, removeFromDriver))
// 删除一个Block数据，会找到数据所在的slave,然后向slave发送一个删除消息
case RemoveBlock(blockId) =>
removeBlockFromWorkers(blockId)
context.reply(true)
// 从BlockManagerInfo中删除一个BlockManager, 并且删除这个 BlockManager上的所有的Blocks
case RemoveExecutor(execId) =>
removeExecutor(execId)
context.reply(true)
// 用于停止 driver 或 executor 端的 BlockManager
case StopBlockManagerMaster =>
context.reply(true)
stop()
// slave 发送心跳给 master , 证明自己还活着
case BlockManagerHeartbeat(blockManagerId) =>
context.reply(heartbeatReceived(blockManagerId))
// 用于检查 executor 是否有缓存 blocks（广播变量的 blocks 不作考虑，因为广播变量的 block 不会汇报给 Master）
case HasCachedBlocks(executorId) =>
blockManagerIdByExecutor.get(executorId) match {
case Some(bm) =>
if (blockManagerInfo.contains(bm)) {
val bmInfo = blockManagerInfo(bm)
context.reply(bmInfo.cachedBlocks.nonEmpty)
} else {
context.reply(false)
}
case None => context.reply(false)
}
master -> slave
// slave删除自己BlockManager上的一个Block
case RemoveBlock(blockId) =>
doAsync[Boolean]("removing block " + blockId, context) {
blockManager.removeBlock(blockId)
true
}
// 删除Rdd对应的Block数据
case RemoveRdd(rddId) =>
doAsync[Int]("removing RDD " + rddId, context) {
blockManager.removeRdd(rddId)
}
// 删除 shuffleId对应的BlockId的Block
case RemoveShuffle(shuffleId) =>
doAsync[Boolean]("removing shuffle " + shuffleId, context) {
if (mapOutputTracker != null) {
mapOutputTracker.unregisterShuffle(shuffleId)
}
SparkEnv.get.shuffleManager.unregisterShuffle(shuffleId)
}
// 删除 BroadcastId对应的BlockId的Block
case RemoveBroadcast(broadcastId, _) =>
doAsync[Int]("removing broadcast " + broadcastId, context) {
blockManager.removeBroadcast(broadcastId, tellMaster = true)
}
// 获取一个Block的存储级别和所占内存和磁盘大小
case GetBlockStatus(blockId, _) =>
context.reply(blockManager.getStatus(blockId))
case GetMatchingBlockIds(filter, _) =>
context.reply(blockManager.getMatchingBlockIds(filter))
case TriggerThreadDump =>
context.reply(Utils.getThreadDump())
存储
在blockManager被创建的时候创建了MemoryStore和DiskStore两个对象用以存取block。
DiskStore
diskSore就是基于磁盘来存储数据的，diskStore有一个成员DiskBlockManager，其主要作用就是逻辑block和磁盘block的映射，block的blockId对应磁盘文件中的一个文件。
def getFile(filename: String): File = {
// Figure out which local directory it hashes to, and which subdirectory in that
val hash = Utils.nonNegativeHash(filename)
val dirId = hash % localDirs.length
val subDirId = (hash / localDirs.length) % subDirsPerLocalDir
// Create the subdirectory if it doesn't already exist
val subDir = subDirs(dirId).synchronized {
val old = subDirs(dirId)(subDirId)
if (old != null) {
old
} else {
val newDir = new File(localDirs(dirId), "%02x".format(subDirId))
if (!newDir.exists() && !newDir.mkdir()) {
throw new IOException(s"Failed to create local dir in $newDir.")
}
subDirs(dirId)(subDirId) = newDir
newDir
}
}
new File(subDir, filename)
}
通过blockId的hash值和localDirs的个数求余来决定在哪个localDir下创建文件，这里的localDirs是可配置的多个目录，可通过SPARK_LOCAL_DIRS进行设置，多个目录以逗号分割，配置多个目录的目的是可分散磁盘的读写压力。另外spark在每个localDir中创建了64（可通过spark.diskStore.subDirectories配置）个子目录来分散文件，子文件的选择也是通过blockId的hash值来计算的。
在diskStore中的putButes方法就是真正写数据到磁盘的方法：
def putBytes(blockId: BlockId, bytes: ChunkedByteBuffer): Unit = {
put(blockId) { fileOutputStream =>
val channel = fileOutputStream.getChannel
Utils.tryWithSafeFinally {
bytes.writeFully(channel)
} {
channel.close()
}
}
}
def put(blockId: BlockId)(writeFunc: FileOutputStream => Unit): Unit = {
if (contains(blockId)) {
throw new IllegalStateException(s"Block $blockId is already present in the disk store")
}
logDebug(s"Attempting to put block $blockId")
val startTime = System.currentTimeMillis
val file = diskManager.getFile(blockId)
val fileOutputStream = new FileOutputStream(file)
var threwException: Boolean = true
try {
writeFunc(fileOutputStream)
threwException = false
} finally {
try {
Closeables.close(fileOutputStream, threwException)
} finally {
if (threwException) {
remove(blockId)
}
}
}
val finishTime = System.currentTimeMillis
logDebug("Block %s stored as %s file on disk in %d ms".format(
file.getName,
Utils.bytesToString(file.length()),
finishTime - startTime))
}
接收一个blockId和要写的字节数据，通过blockId获取到要写的具体文件并得到对应的文件输出流，将该bytes直接write这个流里，完成写文件。
diskStore还有一个重要的方法getBytes方法，即读磁盘文件的方法，通过blockId获取到对应的磁盘文件，以字节 buffer 的形式返回。
此外还有查询blockId对应文件的大小、是否存在blockId对应的文件、删除blockId对应的文件等方法。
MemoryStore
memorySore是基于JVM的堆内存来存储数据，可以用于存数据的内存大小为：
(Runtime.getRuntime.maxMemory * memoryFraction * safetyFraction).toLong
其中memoryFraction 是可通过配置的一个比例（spark.storage.memoryFraction，默认0.6），safetyFraction是一个安全比例，可通过spark.storage.safetyFraction设置。
MemoryStore内部维护了一个hash map来管理所有的block，以block id为key将block存放到hash map中。
private val entries = new LinkedHashMap[BlockId, MemoryEntry[_]](32, 0.75f, true)
放内存就意味着要有足够的内存来放，不然会导致OOM。
若以blockId 对应的数据以bytes数据的方式存放，则可根据其size大小来判断是否有这么多内存来存入，不够的可以放磁盘，对应的方法是:
putBytes[T: ClassTag](blockId: BlockId, size: Long, memoryMode: MemoryMode, _bytes: () => ChunkedByteBuffer): Boolean若是以blockId 对应的数据通过迭代器的方式写入内存，则无法提前知道其数据大小，这里的做法是逐步展开迭代器来检查是否还有空余内存。如果迭代器顺利展开了，那么用来展开迭代器的内存直接转换为存储内存，而不用再去分配内存来存储该 block 数据。如果未能完全开展迭代器，则返回一个包含 block 数据的迭代器，其对应的数据是由多个局部块组合而成的 block 数据，对应的方法是：
putIteratorAsValues[T](blockId: BlockId, values: Iterator[T], classTag: ClassTag[T]): Either[PartiallyUnrolledIterator[T], Long] putIteratorAsBytes[T](blockId: BlockId, values: Iterator[T], classTag: ClassTag[T], memoryMode: MemoryMode): Either[PartiallySerializedBlock[T], Long]
通过memoryStore读数据也有两种方式，一个是以字节buffer的形式返回指定的block数据，另一个是以迭代器的形式返回指定的block数据。
blockManager对外服务
blockManager典型的几个应用场景如下：
- spark shuffle过程的数据就是通过blockManager来存储的。
- spark broadcast 将task调度到多个executor的时候，broadCast 底层使用的数据存储就是blockManager。
- 对一个rdd进行cache的时候，cache的数据就是通过blockManager来存放的。
- spark streaming 一个 ReceiverInputDStream 接受到的数据也是先放在 BlockManager 中， 然后封装为一个 BlockRdd 进行下一步运算的。
