滴滴从KV存储到NewSQL实战

0.导读

本文讲诉滴滴在分布式Nosql存储Fusion之上构建NewSQL的实践之路。详细描述Fusion-NewSQL的特性，应用场景，设计方案。

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索引。

6.总结

Fusion-NewSQL当前已经现已经接入订单、预估、帐单、用户中心、交易引擎等70个核心业务，总QPS超过200W，总数据超过600TB。

固然，Fusion-New不是一个通用完备的NewSQL方案，而是在已有的nosql数据库基础上，经过对SQL协议的支持以及组合各类组件，构建对一个对外表达的数据库，可是这种方式，能够以最小的开发代价，知足大多数的业务场景，具有较高的投入产出比。

7.后续工做

有限制的事物支持，好比让业务规划落在一个节点的数据能够支持单机跨行事务。 实时索引替代异步索引，知足即写即读。目前已经有一个写穿+补偿机制的方案，在没有分布式事务的前提下知足正常状态的实时索引，异常状况下保证数据索引最终一致的方案。 更多的SQL协议和功能支持。

本文做者: ▬

李 鑫 滴滴 | 资深软件开发工程师

多年分布式存储领域设计及开发经验。曾参与Nosql/NewSQL数据库Fusion，分布式时序数据库sentry，NewSQL数据库SDB等系统的设计开发工做。