十三天精通超大规模分布式存储系统架构设计——浅谈B站对象存储(BOSS)实现（上）

本期作者

司春峰

哔哩哔哩技术专家

负责B站对象存储、NoSQL、数据传输、数据库代理等方向的研发工作，致力于提供稳定可靠高性能的存储服务。

背景

BLOB(binary large object)存储，通常也被称为对象存储(OSS, object storage service)。一般用来存储文件，如视频文件、音频文件等。目前，各个云计算厂商都对外提供对象存储服务，其中以亚马逊的S3系统最著名，S3系统也成为行业的事实标准。各个云计算厂商推出的对象存储服务，也纷纷兼容S3标准。

B站由于其内容的独特性(视频网站)，对象存储也有着非常多的需求。下面我们会介绍B站的对象存储的设计与实现。为了便于大家理解，会采用由简单到复杂的过程进行。我们称这个对象存储系统为BOSS(Bilibili Object Storage Service)。目标13天精通超大规模分布式对象存储系统的架构与设计。

Day1

先研读S3协议，我们会发现S3其实提供的接口非常简单，主要有PUT/GET/DEL/LIST等几种类型的接口(这里先忽略分段上传等接口)。
各个接口对应的语义和term定义如下：

• bucketName：bucket的名称，可以理解为目录名
• objectName: 对象名称，即文件名
• PUT: 以 bucketName + objectName 作为key(主键)，上传数据(比如一段视频)
• GET: 以 bucketName + objectName 为key，下载数据(比如刚刚上传的视频)
• DEL: 以 bucketName + objectName 为key，把之前上传的视频进行删除
• LIST: 以bucketName和指定的前缀为过滤条件，列出符合条件的key。比如以bucketName为前缀，列出已经上传的文件名 列表(根据上面的例子，就是列出已经上传的视频文件名)

看到这里，假设我们有一个性能非常强大的SQL服务(比如Mysql ORZ), 那么我们可以建一张表(比如表名为data_and_meta)，然后将数据存入即可。

表结构

系统架构

此时我们的系统架构如下图所示

• 协议接入层(解析http请求，并转换为对MySQL的操作)
• 存储层(MySQL服务)

其中协议网关可以采用golang编写，解析S3协议，并转换为后端对MySQL服务的操作即可。至此，Day1的工作完成， 收工回家。

Day2

第一天在设计表结构的时候，数据和元数据都堆在一起，相互有影响。今天我们将元数据和数据拆开来。新建两张表，meta表和data表。

meta和data分离的表结构

如上图所示，通过object_id将meta表和数据进行关联，object id为一个int64类型(先不考虑object_id哪里来的)

优点

1. 元数据和数据进行了分离存储
2. 可以实现数据和元数据的分别更新(可以预见，后面需要新增字段，满足不同需求)
3. 甚至可是实现rename操作(当然S3没有这个需求)

问题

• 对于相对比较大的文件，性能有可见的问题(先不考虑分段上传)。假设上传一个100MB的文件，只能放在MySQL的一个字段里面， 需要继续优化

引入Block表

针对单个data字段中的数据太长的问题，我们继续拆分，改成3张表，如下图所示

• name表
• object表
• 数据表(多张)

变化

• object id不再直接指向数据，而是数据所对应的blockID列表。
• blockID列表由一个protobuf进行压缩。(为了便于后续的扩展，我们把数据表的表名也放在pb结构中，这样可以使用一张或多张数据表)
• 通过blockid可以在相应的数据表中，查找到每个block所对应的数据。
• blockId和objectID均为int64类型。

写入流程

1. 接入层接受到数据之后，将数据切割为若干个block，每个block分配一个id(先不管哪里来的)。以blockid为主键，将数据写入到data表中
2. 分配一个object_id(先忽略从哪里分配)，将步骤1中的blockid和数据表的表名压缩到一段pb中，然后以object_id为主键，将这段pb写入到object表中
3. 以bucketName和s3文件名组成主键，将object_id写入到name表中

读取流程

1. 读取流程和写入流程相反
2. 以bucketName和s3文件名为key，从name表中查找到对应的objectid
3. 以objectid为key，从object表中查找到对应的block列表
4. 根据block列表中的长度信息，根据用户读请求的区间，计算出需要读取哪些blockid
5. 以blockid为key，从data表中读取对应的数据，并进行截断和拼装，返回结果。(假设上传一个10MB的文件，1MB一个block，指定读取区间 [1MB+1B, 2M+100KB], 则需要读取第1和第2个block，并进行截断)

新的架构

如上图所示，我们实现了数据存储和元数据存储的分离，并将元数据分为两块，用于处理大文件的场景。

objectId和blockID 哪里来?

回顾上文对objectId和blockID的使用，我们发现这两个ID的用途在于"唯一标识一段数据，不能有重复"。
我们可以有几种产生ID的方法:

1. mysql的自增ID
2. value的CRC
3. 其他外部的ID分配器

这里我们先使用mysql的自增ID作为object id。使用CRC的问题在于，同一个文件两次上传(使用不同的文件名)，以block的CRC和整个的CRC分别作为blockid和objectid，会导致在data表和object表中，只有一条记录(CRC相同)，删除的时候会导致实际的数据被删除(当然我们可以使用CRC做去重，这里先不讨论)。
Day2工作完成，我们已经实现了元数据和数据分离存储。

Day3

目前我们接入层直接访问MySQL来进行数据和元数据的读写操作，和MySQL耦合比较严重。今天我们在MySQL和网关之间加一层，用于屏蔽存储层的具体实现。

新架构

网关与S3元数据服务和IO服务之间走RPC进行通信(比如gprc/brpc)，S3元数据服务和IO服务通过SQL访问后端MySQL, 用于屏蔽后端具体的存储实现。

Day4

回顾过去两天的工作，我们已经实现了元数据和数据的分离存储。但是作为一个分布式存储系统，我们离高可用、高可靠、水平扩展能力，差距很大，需要继续改进。
数据存储部分通常采用sharding的方式进行集群化(也就是进行分片)。而sharding的方式， 又分为一次映射和二次映射的方式。

• 一次映射: 指将对应的key(比如我们这里的object id)直接取模，然后落到对应的后端服务器上(物理存储节点，比如某个MySQL中)。
• 二次映射: 可以理解为先将用户的key 映射到一段抽象(虚拟)的结构上，然后再将这个抽象的数据结构映射到实际的存储节点上。

一次映射过程

1. 当路由层收到PUT请求时， 此时输入为(key, value)，其中key为int64类型(blockid)。
2. 假设后端有4台MySQL服务器， key % 4, 得到对应MySQL服务器的下标。
3. 将请求发送给对应的MySQL服务器即可。

一次映射的缺点

1. 没有办法做扩容和缩容操作(比如4台MySQL变成5台之后，按照5进行取模操作，MySQL server对应的下标会发生变化)
2. 替换节点(比如MySQL所在物理机坏了)，必须做一一对应，否则下标无法对应。

一个简单的改进

1. 路由层收到请求之后，查询MySQL(某种元数据服务，这里简化成MySQL), 来确定与后端的哪台服务器进行通信。
2. 根据MySQL返回的地址信息，与数据层的服务节点(data-1...data-4)进行同行。

缺点

1. 存储路由信息MySQL会成为瓶颈，性能和空间都会有问题。
2. 后端进行扩缩容的时候，需要对路由MySQL进行大量的变更操作(需要修改每条key所对应的存储层服务器的地址信息)。

问题关键

上面的方案的主要的问题在于，路由元数据的压缩不够明显，每条记录的元数据都进行了存储。常识告诉我们，计算和存储之间可以进行转换，即通过计算来降低存储空间。

改进方案

1. 引入虚拟的sharding层
2. 将key通过计算的方式得到一个虚拟的shard
3. 路由MySQL中只存放虚拟的shard到存储层服务器地址的映射信息
4. 虚拟的sharding的数量可以先拍脑袋决定，比如2233个。

新写入过程

1. 路由层收到key之后， hash(key) % 2233 得到一个[0, 2232]之间的数字，比如X， 即shardX。
2. 查询路由MySQL，得到shardX所对应后端的数据存储层MySQL服务器的地址。
3. 与数据存储层MySQL进行通信，获取数据。

优点

1. 这里路由层MySQL只需要存储2233条记录即可，每条为(shardID, 数据存储层MySQL的IP)，shardID为主键。
2. 由于这些数据变动的概率非常小，变动的内容有限，完全可以在路由层的内存中缓存。
3. 存储层进行扩容(缩容操作)的时候，只需要在路由MySQL中修改一条记录即可。即shardID到数据层MySQL新的地址信息。
4. 路由层在感知(主动/被动)到变化后，只需要在更新本地的路由表即可。
5. key(blockid)到shard的映射信息，由计算得到，无需存储(eg: md5sum(blockid) % shard的数量)。

对存储层的影响

假设水平扩容的实现过程如下：

1. 从存储节点A上，将隶属于shardX的所有数据copy到存储节点B上
2. 更新路由表中shardX的地址信息(从节点A的IP变更为节点B的IP)
3. 由于在copy的同时，仍然有数据继续写入，因此需要一些容错逻辑，这里不展开。

写入流程如下：

• IO层将请求发送给存储层节点时，需要标示对应的shard信息(比如shard id)
• 存储层的服务节点在处理请求时
• 判断对应的shard在本地是否存在，存在则进行处理。不存在，返回特定错误，提醒写入逻辑更新路由表(有可能shard在进行迁移)。
• 如果对应的shard存在，这直接写入。

data表schema更新

为了便于迁移属于同一个shard的所有数据(快速扫描出来，该shard的所有数据)，存储层表的schema更新如下(新增shard_id字段)：

block_id	value	shard_id
blockID	实际数据	shardID

迁移数据时，根据shard_id字段进行过滤即可。
至此，Day4工作完成，今天我们完成了数据层的sharding过程，并为水平扩容打下了基础。

Day5

到目前后端的数据均存在MySQL中，MySQL的好处在于稳定易用，但是功能过于复杂，性能也不能满足要求。今天我们对MySQL进行替换。

数据存储节点语义

1. 对外提供PUT/GET 接口(先忽略Del接口)。
2. 相应的参数为shardID, key(int64), value。

根据这些需求，可以将存储节点进行如下两种设计：

存储节点设计简介

1. 从上至下分为 RPC 层、shard层和引擎层
2. RPC层负责通信
3. shard层将RPC请求转换为对具体的某个shard的读写操作
4. engine层则负责将请求转换为对磁盘的读写操作

方案对比

方案1

1. 请求进入RPC层之后， 根据shardid 进行分发，获取到对应的shard实例(句柄)
2. shard使用key和value操作engine层
3. 一个节点(或者一块磁盘)公用一个engine，使用shardID作为key前缀，用于区分不同的shard(在迁移的时候，可以使用shardID为前缀扫描属于该shard的所有的key和value)

方案2

1. 方案2作为方案1的简化版本
2. 区别在于，一个shard实例(句柄)对应一个engine实例(而非方案1的全局公用)
3. 优点在于：

• 实现更简洁
• 进行数据搬迁的时候，可以对整个engine进行snapshot拷贝即可(无需逐条扫描)

引擎的实现

今天我们直接使用RocksDB作为我们的单机引擎，不做其他优化。

新架构

数据存储节点替换了原有的MySQL服务，今天的目标达成，收工。

Day6

前面几天已经实现了数据存储层的sharding。但是sharding只能解决水平扩展问题，容灾仍然有问题。今天我们对数据存储集群的资源重新进行整理。

1. 引入资源池和可用区(故障隔离域)的概念。同一个资源池内的机型同构(简化资源调度逻辑，比如相同的磁盘数量和磁盘大小)。不同的业务可以使用不同的资源池，做到存储层资源隔离。
2. 将不同交换机下的节点定义为不同的可用区(故障隔离域)。可用区之间实现交换机级别的隔离。
3. 每个存储集群由一个或者多个资源池组成。资源池之间IO隔离，资源池内部机型同构。
4. 每个资源池内部，由多个可用区组成。每个可用区由若干台服务器(存储节点)组成。
5. 修改路由表中shard到IP的映射关系。

• 一个shard对应到多个Replica(比如3副本)
• 路由表中存放每个Replica所在存储节点的地址信息。
• 一个shard对应的Replica被放置于不同的可用区中(比如3个Replica放在不同的可用区)。

3副本模式的时候，任何一个交换机下的节点宕机，都不会影响读写操作。

新架构

下图为一个资源池+4可用区的模式，每个shard拥有3个副本(Replica)。

• 资源池A由4个可用区组成(可用区0、可用区1、可用区2、可用区3)。
• 每个可用区由3个存储服务节点(Node)组成。
• 以Shard1为例，其所对应的3个Replica(Replica0,Replica1,Replica)分布于资源池A的3个存储节点上。
• 这3个节点分别位于可用区0，可用区2和可用区3

路由层的MySQL中存储的信息如下：

• shard1->(Replica1, Replica2, Replica3)
• Replica1->IPof(可用区0，Node1)
• Replica2->IPof(可用区2，Node0)
• Replica3->IPof(可用区3，Node2)

集群拓扑信息(路由规则)

如何将一张表对应的shard(Replica)分配到这些存储节点上的呢？

1. 假设一张表(table1)，有2233个shard，每个shard 3个副本。
2. 资源池A有4个可用区，每个可用区3个节点。
3. 则每个节点拥有约558个Replica( (2233 * 3) / (4可用区 * 3节点) = 558.25)。
4. 假设每个节点有30块磁盘，则每个磁盘上有18个replica(558.25/30).
5. 后端集群在创建表(table1)的时候，可以对每个shard逐个处理

• 为该shard对应的3个replica(也可以是多个)，从4个可用区中，挑选出合适的节点
• 在从这些节点上挑选出合适的磁盘
• 将这些映射信息，固话到路由层的MySQL数据库中即可
• 挑选策略可以采用负载最低(Replica数量最少)、随机挑选等多种方式

通过这些策略，最终确保每个存储节点的每块磁盘上的Replica数量大致相同，保证读写压力能够均匀的分摊到后端的所有节点上。

读写模型

回顾上文，在引入多副本之前，一条(K,V)请求到达路由层之后，经过sharding和路由表查找之后，请求会转发给存储层的某一个节点，由该节点进行实际的读写操作。在引入多副本的概念之后，引入了副本间的一致性问题。

一致性问题

1. 假设一个shard 有3个副本，R1，R2，R3
2. 两个client对同一条Key，分别进行了写入(key, value1) 和 (key, value2)
3. 根据请求到达的先后顺序，3副本上最终存储的数据可能状态有2x2x2=8种

如上图所示，只有状态1和状态8的3副本处于一致性状态

对于这种一致性问题，通常我们采取选一个主节点(主角色)，由leader来进行冲突的解决(比如RAFT协议) 如下图所示，对于有主系统，常见的复制模型如下

1. HDFS的pipepine方式
2. RAFT协议的Quoram方式

有主系统(有主角色)的主要问题在于，主节点不可避免的会遇到一些抖动(CPU/磁盘IO)问题，从而带来用户请求的抖动。
回顾上文，我们这个系统里面的key(blockid)由MySQL的递增ID生成，全局唯一。一条key只会被写入一次，不存在写冲突问题。因此我们可以采用最简单星型写入模型。
如下图所示，一条记录由IO服务直接并行写入到多个存储节点，多数成功即可。

优点

1. 可用性高：没有选主过程(异常宕机后，RAFT的选主周期在秒级别，比如5s左右)
2. 长尾小
1. 写入时，任意2个节点ack即可
2. 读取时，任意节点返回即可。同时由于没有更新(覆盖)操作，可以使用backup request规避长尾

副本间的一致性

由于采用大多数成功即可的写入模型， 运行一段时间后，同一个shard的各个副本之间的数据分布如下图所示：

可以看到，不同的Replica之间key的数量有差别， 比如key2在Replica2上不存在，key3在Replica3上不存在。为了解决这些差别，我们引入修复模块(先不考虑Delete的情况)，用来修复不同副本之间的差异。

修复工作流程：

1. 修复模块定期的扫描集群中的所有的shard
2. 对每个shard，获取shard的所有的replica的key列表
3. 计算这些Replica间的diff，将缺失的key读取出来，并进行回填

新的架构

今天的架构演化到如下状态：

总结一下今天的工作：

1. 对后端的存储节点重新进行了划分，引入资源池和可用区的概念
2. 引入副本的概念， 一个shard由多个副本组成，这些副本分布在不同的可用区。
3. 对一个shard的写入，实际上转化为对多个副本的写入，这些副本只需要写入到大多数成功即可。
4. 加入离线修复模块，用于修复一个shard多个副本间的数据不一致问题。

到这里，前六天的内容已经介绍完毕，下一篇我们再来聊聊后七天的工作，欢迎持续关注！

---