我们为什么放弃了TiDB，选择自研NewSQL_数据库_李鑫

点击围观！腾讯 TAPD 助力金融行业研发提效、敏捷转型最佳实践！了解详情 



 写点什么

1. 背景

Fusion-NewSQL 是由滴滴自研的在分布式 KV 存储基础上构建的 NewSQL 存储系统。Fusion-NewSQ 兼容了 MySQL 协议，支持二级索引功能，提供超大规模数据持久化存储和高性能读写。

我们的问题

滴滴的业务快速持续发展，数据量和请求量急剧增长，对存储系统等压力与日俱增。虽然分库分表在一定程度上可以解决数据量和请求增加的需求，但是由于滴滴多条业务线（快车，专车，两轮车等）的业务快速变化，数据库加字段加索引的需求非常频繁，分库分表方案对于频繁的 Schema 变更操作并不友好，会导致 DBA 任务繁重，变更周期长，并且对巨大的表操作还会对线上有一定影响。同时，分库分表方案对二级索引支持不友好或者根本不支持。鉴于上述情况，NewSQL 数据库方案就成为我们解决业务问题的一个方向。

开源产品调研

最开始，我们调研了开源的分布式 NewSQL 方案：TIDB。虽然 TIDB 是非常优秀的 NewSQL 产品，但是对于我们的业务场景来说，TIDB 并不是非常适合，原因如下：

我们需要一款高吞吐，低延迟的数据库解决方案，但是 TIDB 由于要满足事务，2pc 方案天然无法满足低延迟（100ms 以内的 99rt，甚至 50ms 内的 99rt）
我们的多数业务，并不真正需要分布式事务，或者说可以通过其他补偿机制，绕过分布式事务。这是由于业务场景决定的。
TIDB 三副本的存储空间成本相对比较高。
我们内部一些离线数据导入在线系统的场景，不能直接和 TIDB 打通。

基于以上原因，我们开启了自研符合自己业务需求的 NewSQL 之路。

我们的基础

我们并没有打算从 0 开发一个完备的 NewSQL 系统，而是在自研的分布式 KV 存储 Fusion 的基础上构建一个能满足我们业务场景的 NewSQL。Fusion 是采用了 Codis 架构，兼容 Redis 协议和数据结构，使用 Rocksdb 作为存储引擎的 NoSQL 数据库。Fusion 在滴滴内部已经有几百个业务在使用，是滴滴主要的在线存储之一。

Fusion 的架构图如下：

我们采用 hash 分片的方式来做数据 sharding。从上往下看，用户通过 Redis 协议的客户端就可以访问 Fusion，用户的访问请求发到 proxy，再由 proxy 转发数据到后端 Fusion 的数据节点。proxy 到后端数据节点的转发，是根据请求的 key 计算 hash 值，然后对 slot 分片数取余，得到一个固定的 slotid，每个 slotid 会固定的映射到一个存储节点，以此解决数据路由问题。

有了一个高并发，低延迟，大容量的存储层后，我们要做的就是在之上构建 MySQL 协议以及二级索引。那么如何将 MySQL 的数据格式转成 Redis 的数据结构存储就是我们必须面临的问题，后面会详细说。

2. 需求

综合考虑大多数用户对需求，我们整理了我们的 NewSQL 需要提供的几个核心能力：

高吞吐，低延迟，大容量。
兼容 MySQL 协议及下游生态。
支持主键查询和二级索引查询。
Schema 变更灵活，不影响线上服务稳定性。

3. 架构设计

Fusion-NewSQL 由下面几个部分组成：

解析 MySQL 协议的 DiseServer
存储数据的 Fusion 集群-Data 集群
存储索引信息的 Fusion 集群-Index 集群
负责 Schema 的管理配置中心-ConfigServer
异步构建索引程序-Consumer 负责消费 Data 集群写到 MQ 中的 MySQL-Binlog 格式数据，根据 schema 信息，生成索引数据写入 Index 集群。
外部依赖，MQ，Zookeeper

架构图如下：

4. 详细设计

存储结构

MySQL 的表结构数据如何转成 Redis 的数据结构是我们面临的第一个问题。

如下图：

我们将 MySQL 表的一行记录转成 Redis 的一个 Hashmap 结构。Hashmap 的 key 由表名+主键值组成，满足了全局唯一的特性。下图展示了 MySQL 通过主键查询转换为 Redis 协议的方式：

除了数据，索引也需要存储在 Fusion-NewSQL 中，和数据存成 hashmap 不同，索引存储成 key-value 结构。根据索引类型不同，组成 key-value 的格式还有一点细微的差别(下面的格式为了看起来直观，实际上分隔符，indexname 都是做过编码的)：

唯一索引：

Key:

table_indexname_indexColumnsValue

Value: Rowkey

非唯一索引：

Key:

table_indexname_indexColumnsValue_Rowkey

Value：null

造成这种差异的原因就是非唯一索引在加入 Rowkey 之前的部分是有可能重复的，无法全局唯一。另外，唯一索引不将 Rowkey 编码在 key 中，是因为在查询语句是单纯的“=”查询的时候直接 get 操作就可以找到对应的 Rowkey 内容，而不需要通过 scan，这样的效率更高。

后面会在查询流程中重点讲述如何通过二级索引查询到数据。

数据读写流程

数据写入

用户通过 MySQL-sdk 将协议发给 dise-server
dise-server 根据 schema 对用户写入的 SQL 做校验
dise-server 将校验通过的 SQL 转成 Redis 的 Hashmap 结构，通过 Redis 协议发给 Data 集群
Data 集群将数据写入 wal 文件，并将数据存储 rocksdb。
Data 集群后台线程将 wal 文件消费，转成 MySQL-Binlog 格式。将数据发到 MQ
异步索引模块消费 MQ，将 MySQL-Binlog 根据操作类型（insert，update,delete）配合 schema 信息，构建索引信息，并将索引数据写入 index 集群。

通过上面的链路，用户的一条 MySQL 写操作就完成了数据存储和索引构建。由于通过数据构建索引这一步是通过 MQ 异步完成，所以会存在数据和索引有一定的时间差的情况。

查询

下面是一个使用二级索引查询数据的案例：

dise-server 接收到 SQL 查询，根据条件，选择索引，如果没有命中任何索引，给用户返回错误（Fusion-NewSQL 不能以非索引字段作为查询条件）。
根据选中的索引，构建查询范围，通过 scan 命令遍历 Index 集群，获取符合条件的主键集合。下图以一个 SQL 查询，展示使用 scan 遍历二级索引的例子：

根据主键，通过 hgetall 命令向 Data 集群查询符合条件的结果集。
将结果集构建成 MySQL 的结果返回给用户。

根据上面索引数据的格式可以看到，scan 范围的时候，前缀必须固定，映射到 SQL 语句到时候，意味着 where 到条件中，范围查询只能有一个字段，而不能多个字段。比如：

索引是 age 和 name 两个字段的联合索引。如果查询语句如下：

select * from student where age > 20 and name >‘W’;

scan 就没有办法确定前缀，也就无法通过 index_age_name 这个索引查询到满足条件的数据，所以使用 KV 形式存储到索引只能满足 where 条件中有一个字段是范围查询。当然可以通过将联合索引分开存放，多次交互搜索取交集的方式解决，但是这就和我们降低 RPC 次数，降低延迟的设计初衷相违背了。为了解决这个问题，我们引入了 Elastic Search 搜索引擎，这部分后面会详细说明。

Schema 变更

用户涉及 Schema 变更时，会以工单形式发给管控系统。管控系统审批过后，会将变更请求推给配置中心，配置中心进行安全性检查后，将新的 Schema 写入到存储中，并给各个节点推送变更。

字段变更：

节点接收到推送，更新本地的 Schema。对于历史数据，并不真正去修改数据，而是在查询的时候，根据 Schema 信息匹配字段，如果数据比 Schema 缺失某些字段，就使用默认值代替;如果数据比 Schema 多了字段，就隐藏掉多余字段不展示。

新增索引分为两步处理：

新增索引，历史数据不处理，增量数据立刻走索引构建流程。
通过历史索引构建工具，扫描历史数据，构建新索引的 KV，将历史数据完成索引构建。这里有个优化点，扫描 slave 而不是 master，避免对线上产生影响。

5. 生态构建

一个单独的存储产品解决所有问题的时代早已经过去，数据孤岛是没有办法很好服务业务的，Fusion-NewSQL 从设计的那天起就考虑了和其他存储系统的打通。

Fusion-NewSQL 到其他存储系统

Fusion-NewSQL 通过兼容 MySQL 的 Binlog 格式，将数据发到 MQ 中。下游各个系统凡是能接入 MySQL 数据的，都可以通过消费 MQ 中相同格式的 Fusion-NewSQL 数据，将数据存到其他系统中。这样的方式用最小的工作量最大程度做到了兼容。

Hive 到 Fusion-NewSQL

Fusion-NewSQL 还支持将离线的 Hive 表中的数据通过 Fusion-NewSQL 提供的 FastLoad（DTS）工具，将 Hive 表数据转入到 Fusion-NewSQL，满足离线数据到在线的数据流动。

如果用户自己完成数据流转，一般会扫描 Hive 表，然后构建 MySQL 的写入语句，一条条将数据写入到 Fusion-NewSQL，流程如下面这样：

MySQL-client 将写请求发给 DiseServer。
DiseServer 将 MySQL 写做解析，转成 hashmap 将转换后的数据以 Redis 协议发给 Data 集群。
Data 集群的存储节点收到数据，将数据写到 wal 文件。
Data 集群的存储节点走 Rocksdb 的写流程，这里包括了写 memtable，还有可能 memtable 写满，发生 flush 以及触发后台的 compact。
异步线程消费 wal，将数据构建 MySQL-Binlog 格式发到 MQ。
异步索引程序消费 MySQL-Binlog，构建 Index 集群需要的数据，向 Index 集群发送写入请求。
Index 集群的存储节点写 wal。
Index 集群的存储节点进入 Rocksdb 的写流程。

从上面的流程可以看出这种迁移方式有几个痛点：

有这种 Hive 到 Fusion-NewSQL 数据导入需求的用户都需要开发一套相同逻辑的代码，维护成本高。
每条 Hive 数据都要经过较长链路，数据导入耗时较长。
离线平台的数据量大，吞吐高，直接大幅提升在线系统的 QPS，对在线系统的稳定性有较大影响。

基于上述的痛点，我们设计了 Fastload 数据导入平台，通过约定 Hive 到 Fusion-NewSQL 的表格式，使用 Hadoop 并发处理数据，并构建 Rocksdb 能识别的 sst 存储文件，绕过复杂的 DISE 写链路，直接将数据导入到 Fusion-NewSQL 中，流程如下：

用户填写工单，选中将指定 Hive 表的某些字段映射为 Fusion-NewSQL 表的字段（这里可以 Hive 中多个字段组成一个 Fusion-NewSQL 字段）。
Hadoop 遍历 Hive 表，并且通过 Zookeeper 获取数据应该存放在 Data 集群和 Index 集群的路由信息
通过上面的遍历，计算，之后，将数据直接构建成、Rocksdb 能识别的 sst，并且其中存的数据已经是按 DISE 的表结构信息组成的 KV 数据。
将 sst 文件直接发送到指定的存储节点，存储节点或通过 Rocksdb 提供的 ingest 功能，直接将 sst 文件加载到 Fusion-NewSQL 中，用户可以读到。

这个方案避免了冗长复杂的写链路，同时不会增加系统的 QPS，在磁盘和网络 IO 没有达到瓶颈的情况下对线上访问几乎是没有任何影响；同时，用户只需要填写 Hive 到 Fusion-NewSQL 的 Schema 映射关系即可，不必再关心实现。

通过 Elastic Search 实现复杂查询

在业务使用 MySQL 或 Fusion-NewSQL 的过程中，我们发现有这样一种场景：业务的查询条件很复杂，涉及的字段数，条件，聚合都比较多，这种场景下，业务会选择将 Elastic Search 作为 MySQL 或 Fusion-NewSQL 的下游，将数据导入 Elastic Search，然后通过 Elastic Search 丰富的搜索能力，先从 Elastic Search 中获取数据在 MySQL 或 Fusion-NewSQL 的主键，然后再根据主键获取全部数据。

根据上面的场景，Fusion-NewSQL 提供一个特殊的索引类型：ES。用户在创建索引的时候，可以将需要做复杂查询的字段勾选出来，共同构建成一个 ES 索引，这样既满足了业务需求，避免了每个业务都需要开发一套和 Elastic Search 交互的复杂逻辑，又统一了数据库使用接口都为 MySQL。同时，还弥补了前面提到的 Fusion-NewSQL 的 KV 二级索引不能支持多个字段范围检索的能力。

架构图如下：

ES 索引只是在上图红 4 处，将 ES 索引中包含的字段信息和主键写入到 Elastic Search 中。在查询时绿 1 如果选中了 ES 类型的索引，就根据 where 条件中涉及的字段，组装成 Elastic Search 的 DSL 语句，从 Elastic Search 获取主键，再从 Data 集群获取。由于 Elastic Search 查询的延迟比较慢，Fusion-NewSQL 可以支持一张表的多个索引采用 KV 索引和 ES 索引并存，对于延迟要求高，查询条件相对简单的使用 KV 索引；对于查询条件复杂，延迟要求不高的使用 ES 索引。