Fusion-NewSQL 是由滴滴自研的在分布式 KV 存储基础上构建的 NewSQL 存储系统。Fusion-NewSQL 兼容了 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 协议以及二级索引。

需求

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

• 高吞吐，低延迟，大容量

• 兼容 MySQL 协议及下游生态

• 支持主键查询和二级索引查询

• Schema 变更灵活，不影响线上服务稳定性。

架构设计

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

1.解析 MySQL 协议的 DiseServer

2.存储数据的 Fusion 集群-Data 集群

3.存储索引信息的 Fusion 集群-Index 集群

4.负责 Schema 的管理配置中心-ConfigServer

5.异步构建索引程序-Consumer 负责消费 Data 集群写到 MQ 中的 MySQL-Binlog 格式数据，根据 schema 信息，生成索引数据写入 Index 集群。

6.外部依赖，MQ，Zookeeper

架构图如下：

技术挑战及方案

1.SQL 表转 Hashmap

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

如下图：

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

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

1.唯一索引：

Key: table_indexname_indexColumnsValue Value: Rowkey

2.非唯一索引：

Key: table_indexname_indexColumnsValue_Rowkey Value：null

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

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

2.数据和索引一致性

因为数据和索引分别存储在不同 Fusion 集群，数据和索引的一致性保证就成了 Fusion-New 系统面临的一个关键点，在没有分布式事务的情况下，我们当前选择了保证数据索引的最终一致性。用户写入数据在数据集群中开启 RocksDB 的单机事物，同时按链接保序，这样数据流入 MQ 的时候就是有序的。异步模块从 MQ 中消费出来再批量写入到索引集群，整个流程就保证的索引数据的构建与数据集群真实的顺序一致。当然，这中间存在一个时间窗口的数据不一致，这个时间取决于 MQ 的吞吐能力。

3.二级索引查询

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

dise-server 会根据用户查询条件和当前所有索引做匹配，找到符合的索引，然后通过 Redis 的 scan 命令，按前缀搜索 index 集群的数据，获取符合条件的主键。

如下图：

通过主键，可以直接到 Data 集群查到相应的数据。

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

比如：

索引是 age 和 name 两个字段的联合索引。

如果查询语句如下：

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

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

架构图如下：

我们建议用户将需要复杂查询的字段设置为 ES 索引，consumer 消费 MQ 的时候将这些字段数据写一份到 ES 中，这样对于对查询条件简单，延迟敏感的查询，使用 Index 集群的数据；对条件复杂，延迟不敏感的查询使用 ES。这样解决了二级索引功能丰富性问题。

4.生态构建

一个单独的存储产品解决所有问题的时代早已经过去，数据孤岛是没有办法很好服务业务的，如何与滴滴现有个各个数据系统打通数据，成了我们必须面对的问题。下面分数据流出到其他系统和从其他系统导入两个方面来阐述 Fusion-NewSQL 的数据流动方案。

4.1. Fusion-NewSQL 到其他存储系统

Fusion-NewSQL 是一个新系统，没办法短时间让各个数据系统为我们做适配。既然 Fusion-NewSQL 已经有了 Schema 信息，那么通过兼容 MySQL 的 Binlog 格式，将 Fusion-NewSQL 在数据链路中伪装成 MySQL，就可以直接使用 Mysql 的下游数据流动链路。这样的方式用最小的工作量最大程度做到了兼容。

4.2.Hive 到 Fusion-NewSQL

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

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

流程如下面这样：

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

1.每条 Hive 数据都要经过较长链路，数据导入耗时较长。

2.离线平台的数据量大，吞吐高，数据导入直接大幅提升在线系统的 QPS，对在线系统的稳定性有较大影响。

从上面的痛点可以看出来，主要的问题是离线数据导入使用了在线系统复杂的 IO 链路。所以如何绕过在线的长 IO 链路，做批量导入就成了解决这个问题的关键。我们设计了 Fastload 数据导入平台，绕过在线 IO 路径

流程如下：

通过 Hadoop 并行计算，将需要导入的 Hive 数据直接构建成 Fusion-NewSQL 能识别的 sst 文件。Fusion-NewSQL 直接将 sst 文件从远端下载到本地，然后使用存储节点通过 Rocksdb 提供 ingest 功能，直接将 sst 文件加载到 Fusion-NewSQL 中，用户可以读到加载到 sst 文件中的数据。通过这样的预先构建 sst 文件，直接文件网络传输和存储引擎直接加载的步骤，就避免了数据导入走在线 IO 复杂流程，解决了稳定性问题，同时将数据导入耗时减少到原来的 1/10。

总结

通过解决上面的技术点，我们用了较小的代价，构建了一个基于 KV 存储的 NewSQL 系统，并且快速将 Fusion-NewSQL 系统接入到滴滴整体的数据链路中。虽然这不是一个完备的 NewSQL 系统，但已经可以满足大多数业务场景的需要，切实实现了 20%工作量满足 80%功能的需求。当前 Fusion-NewSQL 已经接入订单、预估、账单、用户中心、交易引擎等核心业务，总的数据如下图：

后续工作

• 有限制的事物支持，比如让业务规划落在一个节点的数据可以支持单机跨行事务；

• 实时索引替代异步索引，满足即写即读。目前已经有一个写穿+补偿机制的方案，在没有分布式事务的前提下满足正常状态的实时索引，异常情况下保证数据索引最终一致的方案；

• 更多的 SQL 协议和功能支持。

作者介绍：李鑫，滴滴出行资深工程师，曾就职蘑菇街，海康威视。多年分布式存储领域设计及开发经验。曾参与 NoSQL/NewSQL 数据库 Fusion，分布式时序数据库 sentry，NewSQL 数据库 SDB 等系统的设计开发工作。

想了解更多关于架构技术相关专题的同学，可以关注 12 月 6 日在北京·国际会议中心举办的【2019 ArchSummit 北京站】会议

https://archsummit.infoq.cn/2019/beijing/track