分布式时序数据库QTSDB的设计与实现

摘要：奇技指南现有的开源时序数据库只支持单机运行，在面临大量数据写入时，会出现查询慢，机器负载高，单机容量的限制。为了解决这一问题，基础架构团队在单机的基础上，开发了集群版简述是一个分布式时间序列数据库，用于处理海量数据写入与查询。

奇技指南
现有的开源时序数据库influxdb只支持单机运行，在面临大量数据写入时，会出现查询慢，机器负载高，单机容量的限制。
为了解决这一问题，360基础架构团队在单机influxdb的基础上，开发了集群版——QTSDB

QTSDB是一个分布式时间序列数据库，用于处理海量数据写入与查询。实现上，是基于开源单机时序数据库influxdb 1.7开发的分布式版本，除了具有influxdb本身的特性之外，还有容量扩展、副本容错等集群功能。

主要特点如下：

为时间序列数据专门编写的高性能数据存储, 兼顾写入性能和磁盘空间占用；

类sql查询语句, 支持多种统计聚合函数；

自动清理过期数据；

内置连续查询，自动完成用户预设的聚合操作；

Golang编写，没有其它的依赖, 部署运维简单；

节点动态水平扩展，支持海量数据存储；

副本冗余设计，自动故障转移，支持高可用；

优化数据写入，支持高吞吐量；

influxdb架构层次最高是database，database下边根据数据保留时长不同分成了不同的retension policy，形成了database下面的多个存储容器，因为时序数据库与时间维度关联，所以将相同保留时长的内容存放到一起，便于到期删除。除此之外，在retension policy之下，将retension policy的保留时长继续细分，每个时间段的数据存储在一个shard group中，这样当某个分段的shard group到期之后，会将其整个删掉，避免从存储引擎内部抠出部分数据。例如，在database之下的数据，可能是30天保留时长，可能是7天保留时长，他们将存放在不同的retension policy之下。假设将7天的数据继续按1天进行划分，就将他们分别存放到7个shard group中，当第8天的数据生成时，会新建一个shard group写入，并将第 1天的shard group整个删除。

到此为止，同一个retension policy下，发来的当下时序数据只会落在当下的时间段，也就是只有最新的shard group有数据写入，为了提高并发量，一个shard group又分成了多个shard，这些shard全局唯一，分布于所有物理节点上，每个shard对应一个tsm存储引擎，负责存储数据。

在请求访问数据时，通过请求的信息可以锁定某个database和retension policy，然后根据请求中的时间段信息，锁定某个（些）shard group。对于写入的情况，每条写入的数据都对应一个serieskey（这个概念后面会介绍），通过对serieskey进行哈希取模就能锁定一个shard，进行写入。而shard是有副本的，在写入的时候会采用无主多写的策略同时写入到每个副本中。查询时，由于查询请求中没有serieskey的信息，所以只能将shard group内的shard都查询一遍，针对一个shard，会在其副本中选择一个可用的物理节点进行访问。

那么一个shard group要有多少shard呢，为了达到最大并发量，又不过分干扰数据整体的有序性，在物理节点数和副本数确定后，一个shard group内的shard数量是机器数除以副本数，保障了当下的数据可以均匀写入到所有的物理节点之上，也不至于因为shard过多影响查询效率。例如，图上data集群有6个物理节点，用户指定双副本，那么就有3个shard。

整个系统分成三个部分：proxy、meta集群、data集群。proxy负责接收请求，无状态，其前可接lvs支持水平扩展。meta集群保存上面提到的逻辑存储层次及其与物理节点的对应关系，通过raft协议保障元数据的强一致，这里meta信息保存在内存中，日志和快照会持久化到磁盘。data集群是真正的数据存储节点，数据以shard为单位存储于其上，每个shard都对应一个tsm存储引擎。

请求到来的时候，经过lvs锁定一台proxy，proxy先根据database、retension policy和时间段到meta集群查找meta信息，最终得到一个shard到物理节点的映射，然后将这个映射关系转换为物理节点到shard的映射返回给proxy，最后根据这个映射关系，到data集群指定的物理节点中访问具体的shard，至于shard之下的数据访问后边会介绍。

influxdb的查询提供类似于关系数据库的查询方式，展示出来类似一个关系表：measurement，时序数据库的时间作为一个永恒的列，除此之外的列分成两类：

1、field

一类是field，他们是时序数据最关键的数据部分，其值会随着时间的流动源源不断的追加，例如两台机器之间在每个时间点上的延迟。

2、tag

另一类是tag，他们是一个field值的一些标记，所以都是字符串类型，并且取值范围很有限。例如某个时间点的延迟field值是2ms，对应有两个标记属性，从哪台机器到哪台机器的延迟，因此可以设计两个tag：from、to。

measurement展示出来第一行是key，剩下的可以看成value，这样tag有tagkey，tagvalue，field有fieldkey和fieldvalue。

当收到一行写入数据时，会转化为如下的格式：

measurement+tagkey1+tagvalue1+tagkey2+tagvalue2+fieldkey+fieldvalue+time。

如果一行中存在多个field就会划分成多条这样的数据存储。influxdb的存储引擎可以理解为一个map，从measurement到fieldkey作为存储key，后边的fieldvalue和time是存储value，这些值会源源不断追加的，在存储引擎中，这些值会作为一列存储到一起，因为是随时间渐变的数据，将他们保存到一起可以提升压缩的效果。另外将存储key去掉fieldkey之后剩余部分就是上边提到的serieskey。

上边提到，访问请求在集群中如何锁定shard，这里介绍在一个shard内的访问。

influxdb的查询类似于sql语法，但是跟sql语句的零散信息无法直接查询存储引擎，所以需要一些策略将sql语句转换成存储key。influxdb通过构建倒排索引来将where后的tag信息转换为所有相关的serieskey的集合，然后将每个serieskey拼接上select后边的fieldkey就组成了存储key，这样就可以按列取出对应的数据了。

通过对tsm存储引擎中存储key内serieskey的分析，能够构建出倒排索引，新版本influxdb将倒排索引持久化到每个shard中，与存储数据的tsm存储引擎对应，叫做tsi存储引擎。倒排索引相当于一个三层的map，map的key是measurment，值是一个二层的map，这个二层的map的key是tagkey，对应的值是一个一层的map，这个一层map的key是tagval，对应的值是一个serieskey的集合，这个集合中的每个serieskey字串都包含了map索引路径上的measurement、tagkey和tagval。

这样可以分析查询sql，用from后的measurement查询倒排索引三级map获得一个二级map，然后再分析where之后多个过滤逻辑单元，以tagkey1=tagval1为例，将这两个信息作为二层map的key，查到最终的值：serieskey的集合，这个集合的每个serieskey字串都包含了measurment、tagkey1和tagval1，他们是满足当下过滤逻辑单元的serieskey。根据这些逻辑单元的与或逻辑，将其对应的serieskey的集合进行交并运算，最终根据sql的语义过滤出所有的符合其逻辑的serieskey的集合，然后将这些serieskey与select后边的fieldkey拼接起来，得到最终的存储·key，就可以读取数据了。

不带聚合函数的查询：如图，对于一个serieskey，需要拼接众多的fieldkey，进而取出多个列的数据，他们出来后面临的问题是怎么组合为一行的数据，influxdb行列约束比较松散，不能单纯按照列内偏移确定行。Influxdb把serieskey和time作为判断列数据为一行的依据，每一个serieskey对应的多列就汇集为一个以多行为粒度的数据流，多个serieskey对应的数据流按照一定顺序汇集为一个数据流，作为最终的结果集返回到客户端。

带聚合函数的查询：这种方式与上边的查询正好相反，这里是针对聚合函数参数field，拼接上众多的serieskey，当然最终目的都是一样，得到存储key，多个存储key可以读取多个数据流，这些数据流面临两种处理，先将他们按照一定的顺序汇集为一个数据流，然后按照一定的策略圈定这个数据流内相邻的一些数据进行聚合计算，进而得到最终聚合后的值。这里的顺序和策略来自于sql语句中group by后的聚合方式。

多数据流的合并聚合方式，也同样适用于shard之上的查询结果。

对于写入就比较简单了，直接更新数据存储引擎和倒排索引就可以了。

对于访问的整个流程上边都已经提到了，这里整体梳理一下：分成两个阶段，在shard之上的查询，在shard之下的查询。

首先访问请求通过lvs锁定到某个proxy，proxy到meta集群中查找meta信息，根据请求信息，锁定database，retension policy和shard group，进而得到众多的shard。

对于写入操作，根据写入时的serieskey，锁定一个shard进行写入，由于shard存在多副本，需要同时将数据写入到多个副本。对于查询，无法通过请求信息得到serieskey，因此需要查询所有的shard，针对每个shard选择一个可用的副本，进行访问。

经过上边的处理就获得shard到物理节点的映射，然后将其反转为物理节点到shard的映射，返回给proxy，proxy就可以在data集群的某个节点访问对应的shard了。

在shard之下的写入访问，需要拆解insert语句，组合为存储键值对存入tsm存储引擎，然后根据组合的serieskey更新倒排索引。

在shard之下的查询访问，分析sql语句，查询倒排索引，获取其相关的serieskey集合，将其拼接field，形成最终的存储key，进行数据访问。然后将众多数据在data节点上进行shard之上的合并聚合，在proxy上进行data之上的合并聚合。

最终proxy将访问结果返回给客户端。

上边提到influxdb针对shard提供副本容错，当写入数据发送到proxy，proxy将数据以无主多写的形式发送到所有的shard副本。meta集群以心跳的形式监控data节点是否在线，在读取的时候，针对同一shard会在在线的data节点中随机选择一个读取节点进行读取。

在写入时如果一个data节点不可用，则会写入到proxy的一个临时文件中，等网络恢复正常会将这些暂存的数据发送到指定节点。

data集群扩容

当有全新节点加入data集群，目前还不支持自动将现有数据进行迁移，不过也做了些努力，为了使当下写入数据尽快应用到新的节点，在新加入节点的时候，会将当下时间作为当下shard group的结尾时间，然后按照全新的data节点数量新建一个shard group，这样当下数据量马上就能均分到各个data节点，而每个shard group相关的meta信息都存储在meta集群里，因此不会对之前数据的读取造成干扰。

data节点短暂不可用

如果data节点处于短期不可用状态，包括短暂的网络故障后自恢复，或者硬件故障后运维人员干预，最终data节点还存有掉线前的数据，那么就可以以原来的身份加入到data集群。对于写入来说，不可用期间proxy会临时存放此data节点的数据，在data加入集群时会将这部分数据再次发送到data节点，保障数据最终一致。

data节点长期不可用

如果data节点由于一些原因，不能或者不需要以原来的身份加入到集群，需要运维人员手动将原来不可用的data节点下线，那么这台机器可用时，可以以全新的data身份加入到集群中，这等同于集群的扩容。

QTSDB集群实现为：写入时根据serieskey将数据写到指定shard，而读取时无法预知serieskey，因此需要查询每个shard。将整个读取过程切分为两个阶段：在data节点上进行存储引擎的读取以及节点内部多shard的合并聚合，在proxy节点将多个data节点的数据汇总，进行后期的合并聚合，形成最终的结果集返回到客户端。

QTSDB现有的集群功能还有不完善的地方，会在之后的使用中不断完善。

本文为360技术原创文章，转载请务必注明出处及文末二维码，谢谢~

关于360技术
360技术是360技术团队打造的技术分享公众号，每天推送技术干货内容
更多技术信息欢迎关注“360技术”微信公众号