B站分布式KV存储实践

Posted 2023-03-08

参考技术A 在B站的业务场景中，存在很多种不同模型的数据，有些数据关系比较复杂像：账号、稿件信息。有些数据关系比较简单，只需要简单的kv模型即可满足。此外，又存在某些读写吞吐比较高的业务场景，该场景早期的解决方案是通过mysql来进行数据的持久化存储，同时通过redis来提升访问的速度与吞吐。但是这种模式带来了两个问题，其一是存储与缓存一致性的问题，该问题在B站通过canal异步更新缓存的方式得以解决，其二则是开发的复杂度，对于这样一套存储系统，每个业务都需要额外维护一个任务脚本来消费canal数据进行缓存数据的更新。基于这种场景，业务需要的其实是一个介于Redis与MySQL之间的提供持久化高性能的kv存储。此外对象存储的元数据，对数据的一致性、可靠性与扩展性有着很高的要求。

基于此背景，我们对自研KV的定位从一开始就是构建一个高可靠、高可用、高性能、高拓展的系统。对于存储系统，核心是保证数据的可靠性，当数据不可靠时提供再高的可用性也是没用的。可靠性的一个核心因素就是数据的多副本容灾，通过raft一致性协议保证多副本数据的一致性。

分布式系统，如何对数据进行分片放置，业界通常有两种做法，一是基于hash进行分区，二是基于range进行分区，两种方式各有优缺点。hash分区，可以有效防止热点问题，但是由于key是hash以后放置的，无法保证key的全局有序。range分区，由于相邻的数据都放在一起，因此可以保证数据的有序，但是同时也可能带来写入热点的问题。基于B站的业务场景，我们同时支持了range分区和hash分区，业务接入的时候可以根据业务特性进行选择。大部分场景，并不需要全局有序，所以默认推荐hash分区的接入方式，比如观看记录、用户动态这些场景，只需要保证同一个用户维度的数据有序即可，同一个用户维度的数据可以通过hashtag的方式保证局部有序。

整个系统核心分为三个组件:
Metaserver用户集群元信息的管理，包括对kv节点的健康监测、故障转移以及负载均衡。

Node为kv数据存储节点，用于实际存储kv数据，每个Node上保存数据的一个副本，不同Node之间的分片副本通过raft保证数据的一致性，并选出主节点对外提供读写，业务也可以根据对数据一致性的需求指定是否允许读从节点，在对数据一致性要求不高的场景时，通过设置允许读从节点可以提高可用性以及降低长尾。

Client模块为用户访问入口，对外提供了两种接入方式，一种是通过proxy模式的方式进行接入，另一种是通过原生的SDK直接访问，proxy本身也是封装自c++的原生SDK。SDK从Metaserver获取表的元数据分布信息，根据元数据信息决定将用户请求具体发送到哪个对应的Node节点。同时为了保证高可用，SDK还实现了重试机制以及backoff请求。

集群的拓扑结构包含了几个概念，分别是Pool、Zone、Node、Table、Shard 与Replica。

基于不同的业务场景，我们同时支持了range分区和hash分区。对于range场景，随着用户数据的增长，需要对分区数据进行分裂迁移。对于hash分区的场景，使用上通常会根据业务的数据量做几倍的冗余预估，然后创建合适的分片数。但是即便是几倍的冗余预估，由于业务发展速度的不可预测，也很容易出现实际使用远超预估的场景，从而导致单个数据分片过大。

之所以不在一开始就创建足够的分片数有两个原因：其一，由于每一个replica都包含一个独立的engine，过多的分片会导致数据文件过多，同时对于批量写入场景存在一定的写扇出放大。其二，每一个shard都是一组raftgroup，过多的raft心跳会对服务造成额外的开销，这一点后续我们会考虑基于节点做心跳合并优化减少集群心跳数。

为了满足业务的需求场景，我们同时支持了range和hash两种模式下的分裂。两种模式分裂流程类似，下面以hash为例进行说明。

hash模式下的分裂为直接根据当前分片数进行倍增。分裂的流程主要涉及三个模块的交互。

metaserver
分裂时，metaserver会根据当前分片数计算出目标分片数，并且下发创建replica指令到对应的Node节点，同时更新shard分布信息，唯一不同的是，处于分裂中的shard状态为splitting。该状态用于client流量请求路由识别。当Node完成数据分裂以后上报metaserver，metaserver更新shard状态为normal从而完成分裂。

Node
node收到分裂请求以后，会根据需要分裂的分片id在原地拉起创建一个新的分片。然后对旧分片的数据进行checkpoint，同时记录旧分片checkpoint对应的logid。新分片创建完成后，会直接从旧分片的checkpoint进行open，然后在异步复制logid之后的数据保证数据的一致性。新分片加载完checkpoint后，原来的旧分片会向raftgroup提交一条分裂完成日志，该日志处理流程与普通raft日志一致。分裂完成后上报分裂状态到metaserver，同时旧分片开始拒绝不再属于自己分片的数据写入，client收到分片错误以后会请求metaserver更新shard分布。
完成分裂以后的两个分片拥有的两倍冗余数据，这些数据会在engine compaction的时候根据compaction_filter过滤进行删除。

Client
用户请求时，根据hash(key) % shard_cnt 获取目标分片。表分裂期间，该shard_cnt表示分裂完成后的最终分片数。以上图3分片的分裂为例:
hash(key) = 4, 分裂前shard_cnt为3，因此该请求会被发送到shard1. 分裂期间，由于shard_cnt变为6，因此目标分片应该是shard4， 但是由于shard4为splitting，因此client会重新计算分片从而将请求继续发送给shard1. 等到最终分裂完成后，shard4状态变更为Normal，请求才会被发送到shard4.
分裂期间，如果Node返回分片信息错误，那么client会请求metaserver更新分片分布信息。

类似于MySQL的binlog，我们基于raftlog日志实现了kv的binlog. 业务可以根据binlog进行实时的事件流订阅，同时为了满足事件流回溯的需求，我们还对binlog数据进行冷备。通过将binlog冷备到对象存储，满足了部分场景需要回溯较长事件记录的需求。

直接复用raftlog作为用户行为的binlog，可以减少binlog产生的额外写放大，唯一需要处理的是过滤raft本身的配置变更信息。learner通过实时监听不断拉取分片产生的binlog到本地并解析。根据learner配置信息决定将数据同步到对应的下游。同时binlog数据还会被异步备份到对象存储，当业务需要回溯较长时间的事件流的时候，可以直接指定位置从S3拉取历史binlog进行解析。

基于上述提到的binlog能力，我们还基于此实现了kv的多活。learner模块会实时将用户写入的数据同步到跨数据中心的其他kv集群。对于跨数据中心部署的业务，业务可以选择就近的kv集群进行读取访问，降低访问延时。

kv的多活分为读多活和写多活。对于读多活，机房A的写入会被异步复制到机房B，机房B的服务可以直接读取本机房的数据，该模式下只有机房A的kv可以写入。对于写多活，kv在机房A B 都能同时提供写入并且进行双向同步，但是为了保证数据的一致性，需要业务上做数据的单元化写入，保证两个机房不会同时修改同一条记录。通过将用户划分单元，提供了写多活的能力。通过对binlog数据打标，解决了双向同步时候的数据回环问题。

对于用户画像和特征引擎等场景，需要将离线生成的大量数据快速导入KV存储系统提供用户读取访问。传统的写入方式是根据生成的数据记录一条条写入kv存储，这样带来两个问题。其一，大批量写入会对kv造成额外的负载与写入带宽放大造成浪费。其次，由于写入量巨大，每次导入需要花费较长的时间。为了减少写入放大以及导入提速，我们支持了bulk load的能力。离线平台只需要根据kv的存储格式离线生成对应的SST文件，然后上传到对象存储服务。kv直接从对象存储拉取SST文件到本地，然后直接加载SST文件即可对外提供读服务。bulk load的另外一个好处是可以直接在生成SST后离线进行compaction，将compaction的负载offload到离线的同时也降低了空间的放大。

由于LSM tree的写入特性，数据需要被不断的compaction到更底层的level。在compaction时，如果该key还有效，那么会被写入到更底层的level里，如果该key已经被删除，那么会判断当前level是否是最底层的，一条被删除的key，会被标记为删除，直到被compaction到最底层level的时候才会被真正删除。compaction的时候会带来额外的写放大，尤其当value比较大的时候，会造成巨大的带宽浪费。为了降低写放大，我们参考了Bitcask实现了kv分离的存储引擎sparrowdb.

sparrowdb 介绍

用户写入的时候，value通过append only的方式写入data文件，然后更新索引信息，索引的value包含实际数据所在的data文件id，value大小以及position信息，同时data文件也会包含索引信息。与原始的bitcask实现不一样的是，我们将索引信息保存在 rocksdb。

更新写入的时候，只需要更新对应的索引即可。compaction的时候，只需将索引写入底层的level，而无需进行data的拷贝写入。对于已经失效的data，通过后台线程进行检查，当发现data文件里的索引与rocksdb保存的索引不一致的时候，说明该data已经被删除或更新，数据可以被回收淘汰。

使用kv存储分离降低了写放大的问题，但是由于kv分离存储，会导致读的时候多了一次io，读请求需要先根据key读到索引信息，再根据索引信息去对应的文件读取data数据。为了降低读访问的开销，我们针对value比较小的数据进行了inline，只有当value超过一定阈值的时候才会被分离存储到data文件。通过inline以及kv分离获取读性能与写放大之间的平衡。

在分布式系统中，负载均衡是绕不过去的问题。一个好的负载均衡策略可以防止机器资源的空闲浪费。同时通过负载均衡，可以防止流量倾斜导致部分节点负载过高从而影响请求质量。对于存储系统，负载均衡不仅涉及到磁盘的空间，也涉及到机器的内存、cpu、磁盘io等。同时由于使用raft进行主从选主，保证主节点尽可能的打散也是均衡需要考虑的问题。

副本均衡
由于设计上我们会尽量保证每个副本的大小尽量相等，因此对于空间的负载其实可以等价为每块磁盘的副本数。创建副本时，会从可用的zone中寻找包含副本数最少的节点进行创建。同时考虑到不同业务类型的副本读写吞吐可能不一样导致CPU负载不一致，在挑选副本的时候会进一步检查当前节点的负载情况，如果当前节点负载超过阈值，则跳过该节点继续选择其他合适的节点。目前基于最少副本数以及负载校验基本可以做到集群内部的节点负载均衡。
当出现负载倾斜时，则从负载较高的节点选择副本进行迁出，从集群中寻找负载最低的节点作为待迁入节点。当出现节点故障下线以及新机器资源加入的时候，也是基于均值计算待迁出以及迁入节点进行均衡。

主从均衡
虽然通过最少副本数策略保证了节点副本数的均衡，但是由于raft选主的性质，可能出现主节点都集中在部分少数节点的情况。由于只有主节点对外提供写入，主节点的倾斜也会导致负载的不均衡。为了保证主节点的均衡，Node节点会定期向metaserver上报当前节点上副本的主从信息。
主从均衡基于表维度进行操作。metaserver会根据表在Node的分布信息进行副本数的计算。主副本的数量基于最朴素简单的数学期望进行计算: 主副本期望值 = 节点副本数 / 分片副本数。下面为一个简单的例子:
假设表a包含10个shard，每个shard 3个replica。在节点A、B、C、D的分布为 10、5、6、9. 那么A、B、C、D的主副本数期望值应该为 3、1、2、3. 如果节点数实际的主副本数少于期望值，那么被放入待迁入区，如果大于期望值，那么被放入待迁出区。同时通过添加误差值来避免频繁的迁入迁出。只要节点的实际主副本数处于 [x-δx,x+δx] 则表示主副本数处于稳定期间，x、δx 分别表示期望值和误差值。
需要注意的是，当对raft进行主从切换的时候，从节点需要追上所有已提交的日志以后才能成功选为主，如果有节点落后的时候进行主从切换，那么可能导致由于追数据产生的一段时间无主的情况。因此在做主从切换的时候必须要检查主从的日志复制状态，当存在慢节点的时候禁止进行切换。

3.7 故障检测&修复
一个小概率的事件，随着规模的变大，也会变成大概率的事件。分布式系统下，随着集群规模的变大，机器的故障将变得愈发频繁。因此如何对故障进行自动检测容灾修复也是分布式系统的核心问题。故障的容灾主要通过多副本raft来保证，那么如何进行故障的自动发现与修复呢。

健康监测
metaserver会定期向node节点发送心跳检查node的健康状态，如果node出现故障不可达，那么metaserver会将node标记为故障状态并剔除，同时将node上原来的replica迁移到其他健康的节点。
为了防止部分node和metaserver之间部分网络隔离的情况下node节点被误剔除，我们添加了心跳转发的功能。上图中三个node节点对于客户端都是正常的，但是node3由于网络隔离与metaserver不可达了，如果metaserver此时直接剔除node3会造成节点无必要的剔除操作。通过node2转发心跳探测node3的状态避免了误剔除操作。
除了对节点的状态进行检测外，node节点本身还会检查磁盘信息并进行上报，当出现磁盘异常时上报异常磁盘信息并进行踢盘。磁盘的异常主要通过dmesg日志进行采集分析。

故障修复
当出现磁盘节点故障时，需要将原有故障设备的replica迁移到其他健康节点，metaserver根据负载均衡策略选择合适的node并创建新replica， 新创建的replica会被加入原有shard的raft group并从leader复制快照数据，复制完快照以后成功加入raft group完成故障replica的修复。
故障的修复主要涉及快照的复制。每一个replica会定期创建快照删除旧的raftlog，快照信息为完整的rocksdb checkpoint。通过快照进行修复时，只需要拷贝checkpoint下的所有文件即可。通过直接拷贝文件可以大幅减少快照修复的时间。需要注意的是快照拷贝也需要进行io限速，防止文件拷贝影响在线io.

过期数据淘汰
在很多业务场景中，业务的数据只需要存储一段时间，过期后数据即可以自动删除清理，为了支持这个功能，我们通过在value上添加额外的ttl信息，并在compaction的时候通过compaction_filter进行过期数据的淘汰。level之间的容量呈指数增长，因此rocksdb越底层能容纳越多的数据，随着时间的推移，很多数据都会被移动到底层，但是由于底层的容量比较大，很难触发compaction，这就导致很多已经过期的数据没法被及时淘汰从而导致了空间放大。与此同时，大量的过期数据也会对scan的性能造成影响。这个问题可以通过设置periodic_compaction_seconds 来解决，通过设置周期性的compaction来触发过期数据的回收。

scan慢查询
除了上面提到的存在大批过期数据的时候可能导致的scan慢查询，如果业务存在大批量的删除，也可能导致scan的时候出现慢查询。因为delete对于rocksdb本质也是一条append操作，delete写入会被添加删除标记，只有等到该记录被compaction移动到最底层后该标记才会被真正删除。带来的一个问题是如果用户scan的数据区间刚好存在大量的delete标记，那么iterator需要迭代过滤这些标记直到找到有效数据从而导致慢查询。该问题可以通过添加 CompactOnDeletionCollector 来解决。当memtable flush或者sst compaction的时候，collector会统计当前key被删除的比例，通过设置合理的 deletion_trigger ，当发现被delete的key数量超过阈值的时候主动触发compaction。

delay compaction
通过设置 CompactOnDeletionCollector 解决了delete导致的慢查询问题。但是对于某些业务场景，却会到来严重的写放大。当L0被compaction到L1时候，由于阈值超过deletion_trigger ,会导致L1被添加到compaction队列，由于业务的数据特性，L1和L2存在大量重叠的数据区间，导致每次L1的compaction会同时带上大量的L2文件造成巨大的写放大。为了解决这个问题，我们对这种特性的业务数据禁用了CompactOnDeletionCollector 。通过设置表级别参数来控制表级别的compaction策略。后续会考虑优化delete trigger的时机，通过只在指定层级触发来避免大量的io放大。

compaction限速
由于rocksdb的compaction会造成大量的io读写，如果不对compaction的io进行限速，那么很可能影响到在线的写入。但是限速具体配置多少比较合适其实很难确定，配置大了影响在线业务，配置小了又会导致低峰期带宽浪费。基于此rocksdb 在5.9以后为 NewGenericRateLimiter 添加了 auto_tuned 参数，可以根据当前负载自适应调整限速。需要注意的是，该函数还有一个参数 RateLimiter::Mode 用来限制操作类型，默认值为 kWritesOnly,通常情况该模式不会有问题，但是如果业务存在大量被删除的数据，只限制写可能会导致compaction的时候造成大量的读io。

关闭WAL
由于raft log本身已经可以保证数据的可靠性，因此写入rocksdb的时候可以关闭wal减少磁盘io，节点重启的时候根据rocksdb里保存的last_apply_id从raft log进行状态机回放即可。

降副本容灾
对于三副本的raft group，单副本故障并不会影响服务的可用性，即使是主节点故障了剩余的两个节点也会快速选出主并对外提供读写服务。但是考虑到极端情况，假设同时出现两个副本故障呢？ 这时只剩一个副本无法完成选主服务将完全不可用。根据墨菲定律，可能发生的一定会发生。服务的可用性一方面是稳定提供服务的能力，另一方面是故障时快速恢复的能力。那么假设出现这种故障的时候我们应该如何快速恢复服务的可用呢。
如果通过创建新的副本进行修复，新副本需要等到完成快照拷贝以后才能加入raft group进行选举，期间服务还是不可用的。那么我们可以通过强制将分片降为单副本模式，此时剩余的单个健康副本可以独自完成选主，后续再通过变更副本数的方式进行修复。

RaftLog 聚合提交
对于写入吞吐非常高的场景，可以通过牺牲一定的延时来提升写入吞吐，通过log聚合来减少请求放大。对于SSD盘，每一次写入都是4k刷盘，value比较小的时候会造成磁盘带宽的浪费。我们设置了每5ms或者每聚合4k进行批量提交。该参数可以根据业务场景进行动态配置修改。

异步刷盘
有些对于数据一致性要求不是非常高的场景，服务故障的时候允许部分数据丢失。对于该场景，可以关闭fsync通过操作系统进行异步刷盘。但是如果写入吞吐非常高导致page cache的大小超过了 vm.diry_ratio ,那么即便不是fsync也会导致io等待，该场景往往会导致io抖动。为了避免内核pdflush大量刷盘造成的io抖动，我们支持对raftlog进行异步刷盘。

透明多级存储，和缓存结合，自动冷热分离，通过将冷数据自动搬迁到kv降低内存使用成本。
新硬件场景接入，使用SPDK 进行IO提速，使用PMEM进行访问加速。

参考文献
[1] Bitcask A Log-Structured Hash Table for Fast Key/Value Data
[2] Lethe: A Tunable Delete-Aware LSM Engine