时序数据在滴滴实时数据开发平台中的处理和应用-阿里云开发者社区

本文根据演讲视频以及PPT整理而成。

本文将主要围绕以下三个方面进行分享：

背景简介
技术方案
当前现状

一、背景简介
滴滴实时数据开发平台源于这样的初衷，即构建业务监控的双眼，用以监控发生的时事对系统业务交易量是否产生影响。此外，数据可能存在异常和波动，直接监控数据较为低效，时序数据还用以实时报警，帮助相关运营人员排查情况。实时监控和实时报警是滴滴数据管理的两个最核心场景。

2015年以前，滴滴数据管理的架构非常简单。通过扫描Mysql的从表和api数据进行预计算，预计算的操作结果先存储于本地文件，再定时上传到CKV数据库。以计算城市订单量为例，一级目录是城市id，二级目录是日期，预计算的操作结果先按照目录结构存储于本地文件，再定时上传到CKV数据库中。这种简单架构的瓶颈是显而易见的，即指标开发和扩展的难度较大。随着指标增多，在Mysql中维系配置越来越麻烦，并且往往牵一发而动全身。扩展目录也非常麻烦，以增加产品线维度为例，在原来的一级目录和二级目录之间增加产品线维度要求本地文件目录和数据库都需要修改，存储和计算成本呈指数级别增长。问题的根本原因在于，这种架构无法保证数据的及时性和稳定性。链路的延迟以及使用脚本扫描数据使得数据量越大数据查询越慢，数据及时性较差；通过api或者数据库查询无法保证不重不漏地消费数据，数据稳定性较低。

从2015年开始，滴滴针对数据及时性和稳定性两方面进行优化。滴滴数据管理的流程分为几个阶段，首先输入业务数据，对数据进行加工和存储，再经过查询输出相关指标。
2015年以前，数据管理平台使用app请求实现所有连接，不仅影响系统的吞吐性能，还可能存在数据遗漏。2015年之后，滴滴引入了消息队列实现连接，利用消息队列的数据吞吐量优势和ack机制降低了链路时延，同时保证了链路的稳定性。此外，还将数据加工替换为实时计算引擎，进一步降低系统时延。
查询指标涉及到数据计算，数据计算可以在业务数据加工时进行，可以在数据存储时进行，也可以在数据查询时进行。数据计算越迟进行，前期数据的复用率越高，因此滴滴选择在数据查询时进行数据计算。比如，查询实时订单呼叫量时，在数据加工处进行数据计算并存储，在新增实时订单应答量时，仍需开发新的实时任务进行计算；实时集群的资源是有限的，如果每开发一个指标都新增一个实时任务，将会带来极大的资源浪费。假如在数据加工时进行ETL清洗，在数据存储时仅存储简单的订单表，而在数据查询时才进行数据计算就可以复用之前数据计算的结果，从而节省集群资源。这对数据库的查询和存储能力提出了较高的要求。传统的事务型数据库仅针对事务，而不针对数据分析场景，因而不适用于时序数据查询，针对此，滴滴引入了OLAP引擎。由于所有业务数据都具有时序特点，滴滴选择了KAFKA+SMAZA+DRUID的OLAP引擎。

下图展示了2015年到2017年初的架构图，所有线上业务数据使用Kafka进行实时数据流存储，经过实时计算加工，再写回Kafka。然而，这存在开发周期过长，开发语言复杂，现有人力成本跟不上业务需求变更的开发瓶颈。同时，还存在调用链路较长，实时数据资产管理混乱等缺点。为了应对此开发瓶颈，滴滴致力于实现实时数据管理的一体化平台，允许全公司共同参与数据的采集、加工和存储等工作。

由此，滴滴构建了实时计算开发平台Woater，以实现降低开发难度、实时资产管理等优化目标。

二、技术方案
下图展示了2017年至今，滴滴实时计算开发平台采用的技术框架。Mysql的数据从表通过Canal采集，文本日志通过Swan采集，采集到的数据存储于Kafka Topic中。实时计算引擎层引入了Spark Streaming和Flink，对采集到的数据进行数据加工，再写入Druid中。此外，平台还引入了离线数据，使用HIVE将其与实时数据结合，提高了消息队列的数据查询能力。
当数据写入Druid Datasource后，平台会生成所有数据的模板，并落地存储，开放给所有用户使用，开放的功能包括实时监控、报警服务等。此外，平台还允许用户自定义功能，比如根据实时应答率动态调节产品价格等。
另外，滴滴实时数据开发平台还提供了血缘管理、权限管理两个支线功能。血缘管理将平台所有资产，包括Kafka任务和api等实时写入血缘管理模块，方便用户查看任意数据的上下游流动情况。数据开发平台的每个模块都相互解耦，因此滴滴还为每个模块都实现了权限管理功能。

Druid是针对时序数据提供低延时的数据写入以及快速交互式查询的分布式OLAP数据库。在数据写入时，由用户发起实时数据写入任务，通过overload节点将任务发布到各个MiddleManager节点上，MiddleManager节点进而发起子任务，实时拉取数据并存储到Deep Storage中，保证了历史数据的高可用。Druid支持的Deep storage类型包括Amazon S3, HDFS等任意文件系统。在数据查询时，来自客户端的请求首先到达Broker节点，由Broker节点查询全局的数据分片拓扑图，从而将查询拆分成若干子查询，涉及到实时数据的子查询由MiddleManager节点执行，涉及到历史数据的子查询由Historical节点执行，子查询结果在Broker节点汇总并返回给前端。

Druid针对时序数据的存储与查询进行了多项优化。在数据分片设计方面，Druid将所有数据分成了三个属性，包括时间戳，维度和指标。时间戳和指标都采用整数型或者浮点型进行本地存储，因此具有很高的压缩比。字符串数据在本地存储时不会直接存储为字符串格式，而是将其生成映射表。以属性维度的Page为例，映射表可能将“Justin Bieber”映射为0，将“ke$ha”映射为1，那么Page列数据在数据库中将存储为“0011”，因而也具有较高的压缩比。为了加快查询，Druid还为每个值维系一个倒排索引，比如，“Justin Bieber”在Page列的倒排索引为“1100”，“ke$ha”在Page列的倒排索引为“0011”。在查询Page等于“Justin Bieber”时，只需挑选索引中为“1”的项。在进行多列筛选时，只需将各列的索引进行异或计算，比如要查询Page等于“Justin Bieber”且Username等于“Reach”的数据项时，按照倒排索引逻辑，仅需将“0100”和“Justin Bieber”的倒排索引进行异或。

下面列出了Druid的主要优化方法。一方面，Druid采用列式Segment进行存储，另一方面，它采用基于字典的编码对维度列进行压缩，从而提高压缩比。此外，Druid还引入了Rollup聚合存储机制，允许用户自定义数据聚合规则，提高数据的查询效率。
优化后的Druid实现了数据实时可查询，同时提供亚秒级的查询效率，99%的数据查询可以在1秒内返回结果，这对于拥有700+数据源和日均查询量近2000千万的集群而言非常难得。此外，Druid还将数据压缩到了其原始大小的1/30，实现了极高的压缩比。

然而，Druid存在数据加工能力较弱的缺点，仅依赖Druid很难实现两个表的Join操作。
滴滴实时数据开发平台的解决方案是在数据加工层引入主流的SparkStreaming和Flink技术，二者代表了实时计算引擎的两种不同设计方向。Spark Streaming以Spark引擎为基础发展而来，将基于离线数据的批计算压缩为微批操作，使其变成近乎实时的计算。Spark生态具有极高的活跃度，对Spark相关技术熟悉的人员可以快速上手Spark Streaming。Flink适用于低延迟的数据处理场景，支持实时处理数据，同时支持从下至上兼容批处理。此外，Flink还能够提供“exactly once”的实时处理方案和不重不漏的数据消费。

在开发方式方面，滴滴实时数据开发平台提供了多种用户操作以适应更多用户。针对具有流基础的用户，平台提供了Web IDE,支持在线开发编译，在线提交配置，减少了本地配置开发环境的困难。针对具有编程经验的用户，平台提供了StreamSQL和DruidSQL，StreamSQL允许用户使用SQL方式进行数据加工，降低用户的平台上手难度，DruidSQL可用于Druid相关的查询同时帮助用户进行在线调试。此外，滴滴还提供了可视化拖拽等功能帮助指标查询，方便初级用户。