自己的分布式存储系统-BlockManager-–-spark (分布式简介)

教程大全 2025-07-19 20:26:15 浏览次

spark 自己的分布式存储系统 – BlockManager

2018-05-10 09:34:21BlockManager 是 spark 中至关重要的一个组件，在 spark的的运行过程中到处都有 BlockManager 的身影，只有搞清楚 BlockManager 的原理和机制，你才能更加深入的理解 spark。今天我们来揭开 BlockaManager 的底层原理和设计思路，

整体架构

BlockManager 是 spark 中至关重要的一个组件，在 spark的的运行过程中到处都有 BlockManager 的身影，只有搞清楚 BlockManager 的原理和机制，你才能更加深入的理解 spark。今天我们来揭开 BlockaManager 的底层原理和设计思路，

BlockManager 是一个嵌入在 spark 中的 key-value型分布式存储系统，是为 spark 量身打造的，

BlockManager 在一个 spark 应用中作为一个本地缓存运行在所有的节点上，包括所有 driver 和 executor上。 BlockManager 对本地和远程提供一致的 get 和set 数据块接口， BlockManager 本身使用不同的存储方式来存储这些数据，包括 memory, disk, off-heap。

上面是一个整体的架构图， BlockManagerMaster拥有BlockManagerMasterEndpoint 的actor和所有BlockManagerSlaveEndpoint的ref，可以通过这些引用对 slave 下达命令

executor 节点上的BlockManagerMaster 则拥有BlockManagerMasterEndpoint的ref和自身BlockManagerSlaveEndpoint的actor。可以通过 Master的引用注册自己。

在master 和 slave 可以正常的通信之后，就可以根据设计的交互协议进行交互，整个分布式缓存系统也就运转起来了，

初始化

我们知道， sparkEnv 启动的时候会启动各个组件， BlockManager 也不例外，也是这个时候启动的，

启动的时候会根据自己是在 driver 还是 executor 上进行不同的启动过程，

上图是 sparkEnv 在 master上启动的时候，构造了一个 BlockManagerMasterEndpoint, 然后把这个Endpoint 注册在 rpcEnv中，同时也会启动自己的 BlockManager

上图是 sparkEnv 在executor上启动的时候，通过 setupEndpointRef 方法获取到了 BlockManagerMaster的引用 BlockManagerMasterRef，同时也会启动自己的 BlockManager，

在 BlockManager 初始化自己的时候，会向 BlockManagerMasterEndpoint 注册自己， BlockManagerMasterEndpoint 发送 registerBlockManager消息， BlockManagerMasterEndpoint 接受到消息，把 BlockManagerSlaveEndpoint 的引用保存在自己的 blockManagerInfo 数据结构中以待后用。

分布式协议

下面的一个表格是 master 和 slave 接受到各种类型的消息，以及接受到消息后，做的处理。

根据以上的协议，相信我们可以很清楚的猜测整个交互的流程，一般过程应该是这样的， slave的 BlockManager 在自己接的上存储一个 Block, 然后把这个 BlockId 汇报到master的BlockManager , 经过 cache, shuffle 或者 Broadcast后，别的节点需要上一步的Block的时候，会到 master 获取数据所在位置，然后去相应节点上去 fetch。

存储层

在RDD层面上我们了解到RDD是由不同的partition组成的，我们所进行的transformation和action是在partition上面进行的；而在storage模块内部，RDD又被视为由不同的block组成，对于RDD的存取是以block为单位进行的，本质上partition和block是等价的，只是看待的角度不同。在Spark storage模块中中存取数据的最小单位是block，所有的操作都是以block为单位进行的。

BlockManager对象被创建的时候会创建出MemoryStore和DiskStore对象用以存取block，如果内存中拥有足够的内存，就使用 MemoryStore存储，如果不够，就 spill 到磁盘中，通过 DiskStore进行存储。

BlockManager 的 PUT 和GET接口

BlockManager 提供了 Put 接口和 Get 接口，这两个 api 屏蔽了底层的细节，我们来看下底层是如何实现的

blockManager 和 blockTransferService 关系

spark 历史上使用过两套网络框架，最开始的时候， rpc 调用使用的是 akka, 大文件传输使用的是 netty, 后面统一全部使用 netty, 这里的大文件传输其实走的是 netty, 在启动 blockManager的时候会启动一个 blockTransferService 服务，这个服务就是用来传输大文件用的，对应的具体类是 NettyBlockTransferService，这个实例中也会有 BlocakManager的引用，会启动一个 NettyBlockRpcServer的 netty Handler, 也拥有 BlocakManager 的引用，用来提供服务， BlocakManager 根据 BlockId 获取一个 Block 然后包装为一个 ManagedBuffer 对象，

当我们需要从远端获取一个 Block的时候，就需要 blockTransferService 传输大的字节数组，

首先需要从 driver上获取到 Block的真正存储位置，然后调用 blockTransferService 的 fetchBlocks方法，去其他真正存储节点上去fetch数据，会从 client 资源池中获取一个client, 如果是一对一的进行fetch，使用的是 OneForOneBlockFetcher，这个Fetcher 是以 Chunks 为单位分别单独fetch, 每个 Chunks 也就对应一个Block的数据，根据配置，会进行重试直到***重试次数，发送 OpenBlocks消息，里面会包装对应的是哪个 BlockId, 其他节点服务端会根据 BlockId 从 blockManager中拿到数据，然后用来传输，使用的是 netty 的流式传输方式，同时也会有回调函数，

如果是备份的时候同步上传一个 Block, 其他节点服务端会根据，uploadBlock消息中包含的BlockId, 在本地的BlockManager 中冗余存储一份，

ChunkFetch也有一个类似Stream的概念，ChunkFetch的对象是“一个内存中的Iterator[ManagedBuffer]”，即一组Buffer，每一个Buffer对应一个chunkIndex，整个Iterator[ManagedBuffer]由一个StreamID标识。Client每次的ChunkFetch请求是由（streamId，chunkIndex）组成的唯一的StreamChunkId，Server端根据StreamChunkId获取为一个Buffer并返回给Client；不管是Stream还是ChunkFetch，在Server的内存中都需要管理一组由StreamID与资源之间映射，即StreamManager类，它提供了getChunk和openStream两个接口来分别响应ChunkFetch与Stream两种操作，并且针对Server的ChunkFetch提供一个registerStream接口来注册一组Buffer，比如可以将BlockManager中一组BlockID对应的Iterator[ManagedBuffer]注册到StreamManager，从而支持远程Block Fetch操作。

对于ExternalShuffleService(一种单独shuffle服务进程，对其他计算节点提供本节点上面的所有shuffle map输出)，它为远程Executor提供了一种OpenBlocks的RPC接口，即根据请求的appid，executorid，blockid(appid+executor对应本地一组目录，blockid拆封出)从本地磁盘中加载一组FileSegmentManagedBuffer到内存，并返回加载后的streamId返回给客户端，从而支持后续的ChunkFetch的操作。

Partition 与 Block 的关系

我们都知道， RDD 的运算是基于 partition，每个 task 代表一个分区上一个 stage 内的运算闭包， task 被分别调度到多个 executor上去运行，那么是在哪里变成了 Block 呢，我们以 spark 2.11 源码为准，看看这个转变过程，

一个 RDD 调度到 executor 上会运行调用 getOrCompute方法，

如果 Block 在 BlockManager 中存在，就会从 BlockManager 中获取，如果不存在，就进行计算这个Block, 然后在 BlockManager 中进行存储持久化，方便下次使用，

当然获取的时候是先从本地的 BlockManager 中获取，如果本地没有，然后再从 remote 获取，先从 driver 上获取到元数据 Block的位置，然后去到真正的节点上fetch

如果没有，就进行计算，然后根据存储级别，存储到计算节点本地的BlockManager 的内存或磁盘中，

这样RDD的transformation、action就和block数据建立了联系，虽然抽象上我们的操作是在partition层面上进行的，但是partition最终还是被映射成为block，因此实际上我们的所有操作都是对block的处理和存取。

blockManager 在 spark 中扮演的角色

blockManager 是非常非常重要的一个 spark 组件，我们随便举几个例子，你就知道 BlockManager 多重要了，

spark cache 与 spark broadcast task

我随便举两个例子，看看具体 spark cache 和 spark broadcast 调度 task 的时候怎么用的 blockManager的

spark cache

rdd 计算的时候，首先根据RDD id和partition index构造出block id (rdd_xx_xx)，接着从BlockManager中取出相应的block，如果该block存在，表示此RDD在之前已经被计算过和存储在BlockManager中，因此取出即可，无需再重新计算。如果 block 不存在我们可以计算出来，然后吧 block 通过 doPutIterator 函数存储在节点上的 BlockManager上面，汇报block信息到 driver, 下次如果使用同一个 rdd, 就可以直接从分布式存储中直接取出相应的 block

下面看一下源码

如果存储级别不是 NONE类型就会调用 getOrCompute 这个我们已经看过了，里面实际调用 SparkEnv.get.blockManager.getOrElseUpdate 方法，如果 Block 在 BlockManager 中存在，就会从 BlockManager 中获取，如果不存在，就进行计算这个Block, 然后在 BlockManager 中进行存储持久化，方便下次使用，

在 BlockManager 进行存储后，会调用下面的代码把汇报block信息到 driver，

实际上上想 masterEndpoint 的引用发送一条 UpdateBlockInfo消息， master 会把这个 blockId 对应的 location 放在 driver 上，

同样的如果一个 Block已经计算过了，会到 driver 上获取到 location 信息

spark broadcast task

这个调度 task 到多个 task 上面过程代码太多，我就不贴了，直接说一下流程，

blockManager 在 spark streaming 中的应用

测试 blockManager

我们做一个简单的测试，两端代码的区别就是一个进行了cache ，一个没有进行cache。

我们从日志可以观察出来， ***段代码，两个 job 中都从 hdfs 中读取文件，读取了两次，

有以下日志

我们发现在***次读取文件后，把文件 cache 在了 blockManager 中，下一个 job 运行的时候，在本地 BlockManager 直接发现获取到了 block ，没有读取 hdfs 文件，

在 spark ui 中也发现了 cache的 Block，全部是在内存中缓存的，

想成为Java软件工程师有什么要求？

一般Java工程师工作经验在3年以上的，对Java语言熟悉;熟练操作系统、网络、数据结构等，对Spring、ibatis、struts等开源矿机熟悉;熟悉分布式系统设计和应用;熟悉分布式、缓存、消息机制;对数据存储相关知识，如：MySQL、NoSQL等熟练应用者，月薪过万是轻而易举的事。若想拿到月薪过万，你要对职位要求有所了解，根据职位需要在提升个人的能力，加强个人综合能力，具体学习重点如下：1、精通面向对象分析与设计(OOA/OOD)、涉及模式(GOF，J2EEDP)以及综合模式，熟练掌握class，object，interaction、statediagrams;2、熟悉掌握Java核心库，如：collections、serialization、reflection等;3、了解并应用JVM、classloaders、classreflect，以及垃圾回收的基本工作机制等;4、熟悉使用Java数据库技术，如：persistence/ORM构架、TopLink、JDO等;5、熟悉WEB小应用，必须掌握GUI设计的思想和方法，以及桌面程序的SWING，AWT，SWT;6、学习入门轻量级应用程序框架，如：Spring、PicoContainer、Avalon，以及它们的IoC/DI风格(setter、constructor、interfaceinjection);7、熟练掌握JCP、JSR、JavaIDE，也要学习企业级JavaBeans(EJB)以及它们的不同组件模式;8、熟练使用代码生成工具何单元测试体系，如XDoclet、Maven等。