1. 概述

Apache Druid 诞生于一家广告数据分析平台公司，是一个支持海量数据实时分析的分布式数据库系统，适用于大规模实时数据导入，快速查询分析的 OLAP 场景，包括点击流分析，网络服务器性能监控，应用性能指标分析，数字营销分析，实时交易分析等。Druid 具有时序数据库的一些显著特征，支持低延时数据摄取，按照一定的时间颗粒度对数据进行预聚合，能对历史和实时数据提供亚秒级查询，实现高效数据探索分析。

本文先从类 LSM-tree 索引结构，DataSource 与 Segment 数据存储，内存化数据查询几个方面深入理解 Druid 架构设计思想与实时数据处理机制，然后结合实际业务场景阐述 Druid 时序数据处理方法，以及实践过程中遇到的问题与解决方案。

2. Druid 架构设计解析

为实现大规模数据集实时数据处理与即席查询，通常需要在数据的高效摄入与快速查询之间做取舍与权衡。传统关系型数据库如果想在查询时有更快的响应速度，通常会牺牲一些数据写入的性能以完成索引的创建；反之，如果想获得更快的写入速度，往往要放弃一些索引的创建而导致在查询的时候付出较高的性能代价。Druid 通过其独到的架构设计，基于 DataSource 与 Segment 的数据结构，以及一些巧妙的系统实现细节，既达到了高效的数据实时摄入又能提供性能卓越的快速查询。

2.1 架构概览

Druid 使用列存数据存储方式，采用 Lambda 架构分别处理实时数据和批量数据。实时数据处理部分面向写多读少的优化，批处理部分面向读多写少的优化。如图 2-1所示，Druid 总体架构主要包含以下节点：

• 实时节点（Realtime Node）：即时摄入实时数据，以及生成 Segment 数据文件。历史节点（Historical Node）：加载已生成好的数据文件，以供数据查询。
• 代理节点（Broker Node）：对外提供数据查询服务，并同时从实时节点与历史节点查询数据，合并后返回给调用方。
• 协调节点（Coordinator Node）：负责历史节点的数据负载均衡，以及通过规则配置管理数据生命周期。

同时，Druid 集群还包含以下外部依赖:

• 元数据库（Metastore）：存储 Druid 集群的原数据信息，比如 Segment 的相关信息，一般用 MySQL 或 PostgreSQL。
• 分布式协调服务（Coordination）：为 Druid 集群提供一致性协调服务的组件，通常为 Zookeeper。
• 数据文件存储（DeepStorage）：存储 Segment 数据文件，并供历史节点下载。

对于单节点集群可以是本地磁盘，而对于分布式集群一般是 HDFS。从数据流转的角度来看，实时流数据会被实时节点消费，然后实时节点将生成的Segment 数据文件上传到数据文件存储库；而批量数据经过 Druid 集群消费后会被直接上传到数据文件存储库。同时，查询节点会响应外部的查询请求，并将分别从实时节点与历史节点查询到的结果合并后返回。

2.2 类 LSM-tree 索引结构

数据库的数据大多存储在磁盘上，而不论是机械硬盘还是固态硬盘，对磁盘数据的顺序读写速度都远高于随机读写。传统关系型数据库广泛使用 B+树（B+-tree）及其衍生树索引结构，需要较多的磁盘随机读写，并且随着插入的数据不断增多，B+树叶子节点会慢慢分裂，可能导致逻辑上原本连续的数据实际上存放在不同的物理磁盘块位置上，在做范围查询的时候会产生较高的磁盘 I/O，严重影响性能。2008 年 Google关于 Bigtable[3]的论文中引入了日志结构合并树LSM-tree（Log-Structured Merge-Tree）技术，其主要思想是将磁盘看作一个大的日志，每次都将新的数据及其索引结构添加到日志的最末端，以实现对磁盘的顺序操作，并通过将数据文件预排序的方式克服日志结构方法随机读性能较差的问题，从而提高索引性能。LSM-tree 最大的特点是同时使用了两部分类树的数据结构来存储数据，并同时提供查询。其中一部分数据结构（C0 树）存在于内存缓存（通常叫作 memtable）中，负责接受新的数据插入更新以及读请求，并直接在内存中对数据进行排序；另一部分数据结构（C1 树）存在于硬盘上（这部分通常叫作 sstable），它们是由存在于内存缓存中的 C0 树冲写到磁盘而成的，主要负责提供读操作，特点是有序且不可被更改。LSM-tree 的另一大特点是除了使用两部分类树的数据结构外，还会使用日志文件（称为 commit log）来为数据恢复做保障。这三类数据结构的协作顺序一般是：所有的新插入与更新操作都首先被记录到commit log 中——该操作叫作 WAL（Write Ahead Log），然后再写到 memtable，最后当达到一定条件时数据会从 memtable 冲写到 sstable，并抛弃相关的 log 数据；memtable 与 sstable 可同时提供查询；当 memtable 出问题时，可从 commit log 与 sstable中将 memtable 的数据恢复。LSM-tree 的这种结构非常有利于数据的快速写入（理论上可以接近磁盘顺序写速度），但是不利于读，因为理论上读的时候可能需要同时从memtable 和所有硬盘上的 sstable 中查询数据，这样显然会对性能造成较大的影响。

LSM-tree 比较适合那些数据插入操作远多于数据更新删除操作与读操作的场景，Druid 主要是为时序数据场景设计的，而该场景正好符合 LSM-tree 的优势特点，因此Druid 索引架构吸取了 LSM-tree 的思想。Druid 类 LSM-tree 架构中的实时节点（Realtime Node）负责消费实时数据，与经典 LSM-tree 架构不同的是，Druid 不提供日志及实行 WAL 原则，实时数据首先会被直接加载进实时节点内存中的堆结构缓存区（相当于 memtable），当条件满足时，缓存区里的数据会被冲写到硬盘上形成一个数据块（Segment Split），同时实时节点又会立即将新生成的数据块加载到内存中的非堆区，因此无论是堆结构缓存区还是非堆区里的数据，都能够被查询节点（Broker Node）查询。同时，实时节点会周期性地将磁盘上同一个时间段内生成的所有数据块合并为一个大的数据块（Segment）。这个过程在实时节点中称为 Segment Merge 操作，相当于 LSM-tree 架构中的数据合并操作（Compaction）。合并好的 Segment 会立即被实时节点上传到数据文件存储库（DeepStorage）中，随后历史节点通过协调节点从数据文件存储库将新生成的 Segment 下载到其本地磁盘中，并通过分布式协调服务（Coordination）在集群中声明其从此刻开始负责提供该 Segment 的查询，当实时节点收到该声明后也会立即向集群声明其不再提供该 Segment 的查询，接下来查询节点会转从该历史节点查询此 Segment 的数据。而对于全局数据来说，查询节点会同时从实时节点（少量当前数据）与历史节点（大量历史数据）分别查询，然后做一个结果的整合，最后再返回给用户。

Druid 的上述架构特点为其带来了如下显著的优势：

• 类 LSM-tree 架构使得 Druid 能够保证数据的高速写入，并且能够提供比较快速的实时查询，这十分符合许多时序数据的应用场景。
• 由于 Druid 在设计之初就不提供对已有数据的更改，以及不实现传统 LSM-tree架构中普遍应用的 WAL 原则，虽然这样导致了 Druid 不适应于某些需要数据更新的场景，也降低了数据完整性的保障，但 Druid 相对其它传统的 LSM-tree架构实现来说减少了不少数据处理的工作量，因此在性能方面更胜一筹。
• Druid 对命令查询职责分离模式的借鉴也使得其组件职责分明、结构更加清晰明了，方便针对不同模块进行针对性的优化。

2.3 DataSource 与 Segment 数据结构

Druid 在数据摄入之前，首先需要定义一个数据源（DataSource），DataSource 可以理解为 RDBMS 中的表（Table）。如图 2-2 所示，DataSource 的结构包含以下几个方面：

• 时间列（TimeStamp）：表明每行数据的时间值，默认使用 UTC 时间格式且精确到毫秒级别。每个数据集合都必须有时间列，这个列是数据聚合的重要维度，Druid 会将时间很近的一些数据行聚合在一起。另外，所有的查询都需要指定查询时间范围。
• 维度列（Dimension）：维度用来标识数据行的各个类别信息。这些标识主要用于过滤或者切片数据，维度列字段为字符串类型。
• 指标列（Metric）：指标是用于聚合和计算的列。指标列字段通常为数值类型，计算操作通常包括 Count、Sum 和 Mean等。

无论是实时数据消费还是批量数据处理，Druid 在基于 DataSource 结构存储数据时可选择对任意的指标列进行聚合（Roll Up）操作。该聚合操作主要基于维度列与时间范围两方面的情况。

• 同维度列的值做聚合：所有维度列的值都相同时，这一类行数据符合聚合操作。
• 对指定时间粒度内的值做聚合：在 queryGranularity 指定的范围，对时间列值为 1 分钟内的所有行，聚合操作相当于对数据表所有列做了 Group By 操作。Druid 在数据存储时便可对数据进行聚合操作，该特点使得 Druid 不仅能够节省存储空间，而且能够提高聚合查询的效率。

DataSource 是一个逻辑概念，而 Segment 是数据的实际物理存储格式，Druid 正是通过 Segment 实现了对数据的横纵向切割（Slice and Dice）操作。从数据按时间分布的角度来看，通过参数 segmentGranularity 的设置，Druid 将不同时间范围内的数据存储在不同的 Segment 数据块中，即所谓的数据横向切割。这种设计的好处在于：按时间范围查询数据时，仅需要访问对应时间段内的这些 Segment 数据块，而不需要进行全表数据范围查询，这使效率得到了极大的提高。同时，在 Segment 中也面向列进行数据压缩存储，即所谓的数据纵向切割。而且在 Segment 中使用了 Bitmap 等技术对数据的访问进行了优化，在生成索引文件的时候，对每个列的每个取值生成对应的Bitmap 集合。areaId 为 101 对应的 Bitmap 为“1001”，代表第 1 行和第 4 行的 areaId 为“101” 。举一个查询的例子，假设要筛选 areaId=’101’ and subOrderId=’211222002******’的数据，那么只需要把 areaId =’101’对应的 Bitmap“1001”和 subOrderId 对应的 Bitmap “0101“ 进行一个 and/ && 操作，得到结果为“0100”，代表第 1 行满足筛选条件。通过 Bitmap 可以快速定位要读取的数据，加速查询速度。

2.4 内存化数据查询

Druid 历史节点主要是为了提供数据查询服务，历史节点在启动的时候，首先会检查本地缓存中已存在的 Segment 数据文件，然后从 DeepStorage 中下载目前不在本地磁盘上的 Segment 数据文件。无论是何种查询，历史节点都会首先将相关 Segment 数据文件从磁盘加载到内存，然后再提供查询服务。历史节点的查询效率受内存空间的影响很大：内存空间越大，查询时需要从磁盘加载数据的次数就越少，查询速度就越快；反之，查询时需要从磁盘加载数据的次数就多，查询速度就相对较慢。历史节点具有较佳的可扩展性与高可用性。新的历史节点被添加后，会通过Zookeeper 被协调节点（Coordinator Node）发现，然后协调节点自动分配相关的Segment 给它；原有的历史节点被移出集群后，同样会被协调节点发现，协调节点会将原本分配给它的 Segment 重新分配给其它处于工作状态的历史节点。

Druid 使用 Cache 机制来提高查询效率。Druid 提供了两类介质作为 Cache 以供择：

• 外部 Cache，比如 Memcached
• 本地 Cache，比如(查询)代理节点或历史节点的内存作为 Cache。

如果用代理节点的内存作为 Cache，查询的时候会首先访问其 Cache，只有当不命中的时候才会去访问历史节点与实时节点以查询数据。

3. Druid 应用实践

公司早期在 OLAP 场景下涉及实时订单数据处理部分使用了 Druid，本章节主要介绍相关应用细节，以及实践过程中遇到的问题与解决方案。

3.1 Druid 应用架构简介

如图 3-1 所示，一个典型的 Druid 应用架构包含以下数据处理节点：

1)业务生产系统以 Mysql 为主，MySQL 二进制日志 binlog 记录了所有的 DDL 和DML 语句（除了数据查询语句select）,且以严格的时间事件形式记录，保障事务完整和准确。通过 canal 工具解析 MySQL binlog，获取每一笔订单的实时 PAID 状态，将支付订单中相关字段如： 订单号，金额，份数，门提，下单用户等指标信息，写入kafka 集群。

2) Kafka-index-service 是 Druid 自身携带的扩展插件，能够将 kafka 中流式数据，通过 Druid 组件不断的摄取，使用时需要在 common.runtime.properties 文件中的属性 druid.extensions.loadLis 添加 druid-kafka-indexing-service。

创建 Druid 拉取 kafka 实时数据流的 json 代码文件，通过 curl 方式提交命令：curl -X POST -H 'Content-Type: application/json' -d @stream_data.json，http://localhost:8090/druid/indexer/v1/supervisor，通过 Overlord 节点页面 控制台观察对应的数据摄取任务：

3)Druid 在 segment 文件中存储数据指标，segment 文件用时间参数划分分区。在基础设置中，会为每个时间间隔建立一个 segment 文件，该时间间隔能够在granularitySpec 的 segmentGranularity 参数中配置。同时观察 segment 的文件大小，如果超过一定量应该改变时间间隔或者分区数据，而且对 partitionsSpec 中的targetPartitionSize 参数作出调整（推荐优先设置这个参数为 500 万行）。

4)Druid 基于 Apache Calcite 实现 SQL 查询，即通过 Broker 节点进行连接，用pydruid 或者 JAVA，配置 jdbc 连接，同时也可以配置 local curl 方式进行实时 SQL指标计算，将对应的结果集实时写入 redis，供前端界面展示，查询实时数据。

3.2 Druid 应用问题及解决方案

初期在部署应用 Druid 过程中，淌过不少应用坑，时常有意想不到的事件发生， 给开发和运维带来了不小的挑战。随着应用的逐步深入，集群的稳定性和资源的使用率，逐步得到稳定的提升，同时实时数据处理与查询性能达到良好的效果。以下简述应用过程中踩的部分坑，以及对应的解决方案或优化处理。

1)任务参数不合理导致小 Segment 过多

通过 Coordinator 控制台可以查看到对应的 data_source 整体 segment 文件个数和文件大小。其中我们经常发现，经过一天的 kafka 流式摄取，会生成 N 个 segment文件，但以小文件为主，在计算查询时，小文件过多，会影响实时查询的整体性能。

解决方案：让 Druid 在繁重的查询压力下保持良好的操作性能，且应该让segment 文件的大小在 300mb 至 700mb 之间。

i. 调整任务 json 文件参数 partitionsSpec 中的 targetPartitionSize 等，合理生成任务数和文件大小。

ii. 在凌晨或者数据流不大的情况下，启动 compact 任务，对文件进行合并操作，合并操作的参数，满足 segment 最优的原则。

2)不支持精准去重

Druid 在设计之初，为了追求亚秒级的 OLAP 查询分析，对精准去重计算 exactdistinct count 不支持。如遇到准确去重计算时，会满足不了业务需求。目前 druid只提供近似查询，牺牲部分数据的准确性，如 HyperUniques/Sketch: 规则方式摄入时做计算，做近似去重查询。

解决方案：

i. 在实时要求程度不高的情况下，可以通过时间换空间的方式，对需要计算的字段做 group by 查询，然后 count(1) 实现 精准去重。

ii. 对要做精准去重的字段，在摄入时将字段类型改成"type": "thetaSketch" ,在计算时通过调整 SIZE 大小，可以实现精准去重{ "type": "thetaSketch","name":"dist_username", "fieldName": "去重用户","isInputThetaSketch":

false,"size": 1048576 }。

3)近似计算出现浮点数

在 hyperUniqueCardinality 近似计算时，常常会输出浮点数，导致 where 条件过滤计算时， 无法使用 equalTo 进行计算。

解决方案：

此类 计 算 问 题 通 过 在 postAggregations 中 ， 增 加 一 个 JavaScript postaggregator 计算过程，再利用 Math.round 进行四舍五入即可。

4)数据摄入 Boolean 类型字段值为 NULL

字段类型为 boolean 类型，且做为维度层处理，最后摄取入 Druid 结果后变成了NULL。

解决方案：

Druid 本身是无法将 true / false 之类的 boolean 类型作为维度的，可以考虑将"true"/"false"字符串作为维度存入。但是，如果自定义的 Bean 对象中，有String isTimeout = "false"的属性存在，就不能直接使用JSON.toJSONString 进行转换。因为 toJSONString 方法中会识别出"true"/"false"字符串，并将其自动转化为boolean 类型。因此，需要通过 Map<String, Object>将所有字段都存入，然后再调用JSON.toJSONString 方法即可。

5)SegmentGranularity 和 QueryGranularity 的配置问题

QueryGranularity 是查询的最小时间粒度，它主要影响数据摄入过程和查询过程中的 Roll-up， QueryGranularity 越大，Roll-up 以后的数据量越小，查询越快，但是同时也意味着不能查询更细时间粒度的数据。SegmentGranularity 是 Segment 分段的时间粒度，比如 SegmentGranularity 为 hour，那么 Segment 是按照小时分段。通常我们可以设置 SegmentGranularity>=QueryGranularity，但这并不是必须遵守的。例如有一个 DataSource 的 QueryGranularity 为 day，采用实时摄入的 方式，如果遵守上述原则，SegmentGranularity 至少是 day，那么一天之内 Segment 都不能移交给历史节点，导致查询性能下降。

6)内存增量索引持久化阈值设置问题

Druid 运行中的数据首先在内存增量索引中累积，此过程中会进行 Rollup。当达到一定阈值以后，采用异步线程将内存增量索引持久化到磁盘中，持久化后的索引采用倒排和 Bitmap 索引，适合查询，当达到 SegmentGranularity 设定的条件时，将所有 的 持 久 化 索 引 合 并 成 一 个 Segment 并 移 交 。 决 定 持 久 化 阈 值 的 是intermediatePersistPeriod 和 maxRowsInMemory。intermediatePersistPeriod 的含义是间隔多长时间进行持久化，maxRowsInMemory 是内存增量索引中事件条数达到多少以后进行持久化，两个参数满足其一就会触发持久化。如果将参数设置过小，将会过多地进行持久化，将会影响查询性能；如果将参数设置过大，则会使用过多的内存，影响垃圾回收。实际操作中通过系统运行指标监控调整参数设置。

4. 总结

Druid 在架构设计上借鉴了 LSM-tree 及读写分离的思想，并且基于 DataSource 与Segment 数据文件结构与组织方式，通过数据的预聚合，优化存储结构和内存使用，即提供了实时和离线批量数据摄入的能力，又能够在大规模数据集上实现高效实时数据消费与探索，在时序数据应用场景中满足了 OLAP 的核心功能。但 Druid 是一套比较复杂的分布式系统，为实现其完整的集群功能，不仅需要要熟悉 Broker, Historical,MiddleManager，Coordinator 等集群节点进程功能，优化各节点机器配置及对应的JVM 与线程池配置，还要面对其外部依赖 DeepStorage (HDFS), ZooKeeper, MetadataStorage,可能出现直接或间接影响 Druid 有效运行的各种问题，因而整套系统有较高的运维成本。这些都是我们在做技术选型时需要考虑的因素。

参考文献：

[1] https://druid.apache.org/docs/latest/design/architecture.html

[2] http://static.druid.io/docs/druid.pdf

[3] F. Chang, J. Dean, S. Ghemawat, W. C. Hsieh, D. A. Wallach, M. Burrows, T. Chandra, A.

Fikes, and R. E. Gruber. Bigtable: A distributed storage system for structured data. ACM

Transactions on Computer Systems (TOCS), 26(2):4, 2008.