知乎 Druid 集群优化实践

背景

随着业务的发展，Druid 集群规模不断增长；一个美好的傍晚，打开电脑准备看一场球赛，突然，被一阵急促的报警电话，打破宁静，集群查询失败率飙升。。。此时迅速打开集群监控大盘，发现此时集群 Load 已经被跑飞；对于负责集群的同学来说，在集群发展到早中期，应该是大家都会遇到的问题。

本篇文章主要讲述围绕知乎在 Druid 集群建设发展中遇到的问题以及如何提升稳定性为主题展开。通过阅读文章，可以收获到集群从极速发展到不稳定再到稳定运行的治理历程，踩过的每一个坑，希望对大家有帮助。

先介绍一下 Druid 在知乎的应用，截止目前在知乎 Druid 集群现状 节点70+；数据源300+；存储规模 600TB+。

如上图所示，在知乎 Druid 主要是针对两种场景，一种是实时接入，满足实时指标展示及实时分析；另一种是离线导入 Hive 数据，加速Hive数据查询分析，提高分析师工作效率。主要使用的业务类型为 A/B Testing、渠道管理、APM 、数据邮件等业务。

Druid 简介

本小结主要是针对 Druid 初学者，如果你已经很了解 Druid 架构，可以直接阅读下一章节内容,略过此章节。

Druid 的整体架构如上图所示，其中主要有 3 条路线：

1. 实时摄入的过程： 实时数据会首先按行摄入 Real-time Nodes，Real-time Nodes 会先将每行的数据加入到1个 map 中，等达到一定的行数或者大小限制时，Real-time Nodes 就会将内存中的 map 持久化到磁盘中，Real-time Nodes 会按照segmentGranularity 将一定时间段内的小文件 merge 为一个大文件，生成 Segment，然后将 Segment 上传到 Deep Storage（HDFS，S3）中，Coordinator 知道有Segment 生成后，会通知相应的 Historical Node 下载对应的 Segment，并负责该Segment 的查询。
2. 离线摄入的过程：离线摄入的过程比较简单，就是直接通过 MR job 生成 Segment，剩下的逻辑和实时摄入相同。
3. 用户查询过程： 用户的查询都是直接发送到 Broker Node，Broker Node 会将查询分发到 Real-time 节点和 Historical 节点，然后将结果合并后返回给用户。

各节点的主要职责如下：

Historical Nodes

Historical 节点是整个 Druid 集群的骨干，主要负责加载不可变的 Segment，并负责 Segment的查询（注意，Segment 必须加载到 Historical 的内存中才可以提供查询）。Historical 节点是无状态的，所以可以轻易的横向扩展和快速恢复。Historical 节点 load 和 drop Segment 是依赖 ZK 的，但是即使 ZK 挂掉，Historical 依然可以对已经加载的 Segment 提供查询，只是不能再 load 新Segment，drop 旧 Segment。

Broker Nodes

Broker 节点是Druid查询的入口，主要负责查询的分发和 Merge。 之外，Broker 还会对不可变的 Segment 的查询结果进行 LRU 缓存。

Coordinator Nodes

Coordinator 节点主要负责 Segment 的管理。Coordinator 节点会通知 Historical 节点加载新Segment，删除旧 Segment，复制 Segment，以及 Historical 间的负载均衡。

Coordinator 节点依赖 ZK 确定 Historical 的存活和集群 Segment 的分布。

Real-time Node

实时节点主要负责数据的实时摄入，实时数据的查询，将实时数据转为 Segment，将Segment 分配 给Historical 节点。

Zookeeper

Druid 依赖 ZK 实现服务发现，数据拓扑的感知，以及 Coordinator 的选主。

Metadata Storage

Metadata storage（Mysql） 主要用来存储 Segment 和配置的元数据。当有新 Segment 生成时，就会将Segment的元信息写入 metadata store, Coordinator 节点会监控 Metadata store 从而知道何时 load 新 Segment，何时 drop 旧 Segment。注意，查询时不会涉及 Metadata store。

Deep Storage

Deep storage (S3 and HDFS) 是作为 Segment 的永久备份，查询时同样不会涉及 Deep storage。

知乎 Druid 平台架构演进

平台架构 V1.0

早期 Druid 集群架构如上图所示，功能上支持数据的实时摄入和离线批量导入：

1. 实时数据摄入：基于 Traquility 实现了消费 Kafka 数据，实时摄入 Druid，提供实时查询服务
2. 数据离线导入：基于内部的离线调度平台，可视化支持数仓工程师创建离线定时导入任务
3. 数据存储：数据源所有数据均存储在大集群，Historical 节点均为 SSD 磁盘，数据单备份
4. 监控系统：

随着业务的日益增加，当前技术方案出现了如下痛点：

1. 业务查询超时：随着业务的日益需求的日益增长，查询业务并发不断提高，出现高并发下，查询超时，业务间相互影响
2. 数据可用性差：由于此架构下，数据源的 Segment 在 Historical 节点无备份，任何一台 Historical 节点宕机，整个集群中的数据不可查。
3. 实时数据丢失：基于 Traquility 实现的实时数据摄入，存在超出时间窗口后，数据丢失
4. 存储成本增加：由于业务前期，为了提升查询性能，数据均存储在 SSD 磁盘中

由于前期重点放在拓展业务上，导致发展到一定阶段后，出现了集群建设不合理上带来的痛点；基于以上痛点的问题根源分析，Druid 平台 V2.0 应运而生。

平台架构V2.0

平台架构 V2.0 如上图所示，对比早期集群，做了如下改进：

1. 实时数据摄入：采用 Kafka Indexing Service 方案，解决了超出时间窗口丢失数据的问题，实现了 exactly-once ingestion 语意
2. 业务隔离，数据冷热分层：

冷热分层的依据是什么呢？在确定规则前，对于业务查询的场景及查询历史进行了分析，最近一个月的数据查询占比 97% 左右；一个月以外的数据查询占比比较低，使用特点是用户月末或季度末查询数据所使用

如上图所示，核心业务和其他业务进行存储隔离，内部进行冷热分层，具体存储规则制定：

• 采用统一双副本
• 一个月内的数据存储 ：Hot Tier 存储一份；Cold Tier 存储一份
• 一年内的数据：Cold Tier 存储两份

收益：

• 数据双副本：提升了数据可用性，不会因为某个HIstorical节点宕机而不可用
• 业务隔离：提升了查询响应速度，降低了业务查询之间的相互影响
• 冷热数据分层：降低了数据存储成本，集群成本大约降低 40%

3. 监控系统：

主要实现目标是对数据加载和查询行为进行监控，做到问题前，问题中，问题后均可根据监控发现问题；随着问题的不断出现，监控大盘不断进行迭代，到目前为止，现有监控大盘做到了能够及时发现问题，研判问题，核心监控点如下所述：Coordinator 调度监控

作用：数据源离线加载后，长时间无法数据可见查询；出现问题时此指标会长时间不输出。

指标： 此监控指标代表着每次 Coordinator 对数据源 Segment 管理调度周期以及 Assign 到 Historical 的 SegmentRouter 查询并发监控

作用：设定并发查询阈值报警，及时发现高并发，以免持续高并发影响集群查询响应

查询时间实时监控

作用：集群 CPU Load 持续飙高，影响集群查询响应；快速定位是哪个数据源的查询导致

其实监控系统在建设之初，一般按照官方文档，基础的核心监控指标都会展示，由于篇幅有限，在这里就没有展开详细介绍，只是选取了随着系统运行，发现问题不断迭代出的指标，它们具有典型特点：

a. 在问题前做到预警，比如并发查询监控，能够早于业务发现问题，提前干预，降低对业务的影响；

b.在问题中及时止损，比如查询实时使用CPU时间监控，当集群 Load 持续飙高，影响业务查询时，及时定位数据源，人为介入处理。

4. 集群管理系统：

此系统主要功能是管理 Druid 集群，记录节点状态，展示导入任务记录，搜索查询日志，探测查询行为等功能,具体功能展示如下图：

收益：极大了提升了集群管理效率，为问题决策提供了依据

本小结主要讲述了 Druid 平台不断迭代的过程，从 V1.0 随着问题痛点的出现，经过方案调研和具体情况分析，产出 V2.0 集群构建方案并落地，主要从业务隔离，冷热数据，监控系统打造，管理系统建设几个方面优化集群建设，做到了降低集群成本，迅速发现并解决问题，极大的提升了集群稳定性。

稳定性优化

随着完成推动集群架构的演进，解决了大部分问题，集群可用性提升到一个水平，但是问题又来了，就是追求集群的稳定性要稳定在99.9%以上

加载任务 Pending

问题描述：

首先实时加载任务 Segment 一直处于 Hand Off 阶段，无法释放加载 worker 资源，造成新的实时任务无法启动处于 Pending 状态；其次，对于离线加载任务，任务状态显示成功，但是数据延迟查询不可见

问题分析：

首先看一下 Segment 生成后，Assign 到 Historical 的整个事件流,如下图：

如上图所示，关键流程描述如下：

1. Worker 将生成的 Segment 文件写入持久化文件系统 HDFS 或 S3
2. Worker 将 Segment 元信息写入数据库 Mysql
3. Coordinator 定期调度获取 Segment 元信息和 数据源 的规则Rule
4. Coordinator 将需要加载或删除的 Segment 消息同步到 ZK
5. Historical 从 ZK 获取加载或删除 Segment 的消息
6. Historical 从持久化文件系统拉取 Segment 文件
7. Historical 将对应的同步消息从 ZK删除

Coordinator 通过 ZK 与 Historical 进行 Segment Load 信息调度;当 RealTime 实时加载任务所产生的所有 Segment 加载到 Historical 时才能释放Druid MiddleManager 上的 worker;

根据线上worker 日志已经确定1，2两步已经完成；需要进一步分析 Coordinator 日志及调度线程模型明确关键流程中问题：

Coordinator 中负责 Segment 管理的模型是串行处理模型, 因此问题根源性原因是一个调度周期内，产生了大量的Segment（按照经验2W+ Segment)，导致此调度周期长时间运行，在此调度周期内产生的 Segment 无法 Assign 到 Historical,最终导致实时加载任务无法释放 Worker；造成数据积压。

解决方案：

规范业务导入数据量，减少 Segment 个数， 每次导入的数据不超过 7 天。

脏查询治理

还记得开篇中，突然被报警打破的宁静吗？经过 Druid 架构调整后，这种问题不会出现了，但是进一步提高了标准，集群查询可用性要不低于99.9%，因此10分钟持续低于SLA标准，也不能接受。

首先描述一下脏查询的定义：就是某些查询导致集群 Load 高，造成正常的查询不能响应，影响集群的SLA；对于 Druid 所谓的脏查询主要包括：查询无filter，数据全扫描；javascripe 正则匹配查询; 次周或次月留存等。

问题描述：

偶尔集群查询SLA低于 99%,造成部分查询超时

问题分析：

集群查询 SLA 降低监控

CPU Load 监控

查询CPU耗时实时监控

从这个监控大盘定位数据源，经过查询分析，此数据源在进行近一个月内的次月留存分析，这种查询非常消耗计算资源。

根据监控可以得出结论，是由于某些脏查询导致节点资源耗尽，造成集群查询SLA下降。

解决方案：

利用集群管理系统中查询行为探测，细粒度分析了最近几个月的查询行为，发现99.87%的查询都能够秒级返回，脏查询共同的特点是分钟级返回结果；因此制定了一个治理脏查询的策略：在 Broker 层面统一设定查询超时时间，热数据超时时间为 1 Min;冷数据超时时间为 3 Min。

配置：

druid.server.http.maxIdleTime=PT1m
druid.server.http.defaultQueryTimeout=60000
druid.server.http.maxQueryTimeout=60000           

踩坑：如果线上的版本低于0.13.0 需要打 Patch(issues:) defaultQueryTimeout 参数才能生效。

修复 Historical 慢启动

问题描述

目前冷数据存储Historical （12*2T SATA HDD）节点存储超过 50K Segments,大约有10TB+ 的数据规模，每当节点OOM宕机或者重启服务，会耗费半个小时左右重启时间；但带来的后果是造成数据移动，影响实时加载任务，甚至造成数据丢失。

问题分析

根据（issue:）的问题分析及解决方案，通过日志和源码分析是Historical 启动初始化流程，发现Historical 初始化过程需要读取数据源中 Segment 中每一列的元信息， Segment 文件格式如下图所示：

如上图所示，就是需要去磁盘读取每一列的 ColumnDescriptor；可以理解当数据节点存储数据量比较大，又存储在多列的情况下，会产生很多随机IO操作；HDD 本身100左右 极限 IOPS；因此大量的时间耗费在了随机IO操作上。

在现有的数据量下，重启一次耗时在 30 Min以上，其实对于系统的稳定性留下了比较大的隐患。

解决方案

内部版本引入PR ，由于社区中是基于0.13+以上解决的此问题，因此引入此PR 到内部版本，主要解决了接口不兼容的问题；并引入到了0.12.1版本，需要此版本Patch 的可以自取 。

解决的核心思想就是典型的 Lazy 加载模式，如下代码所示：

Map<String, Column> columns = new HashMap<>(); ===》Map<String, Supplier<Column>> columns = new HashMap<>();           

通过 Guava 库中的Supplier特性来实现 Lazy 加载模式，Supplier<Column> 修饰的Column对象，在第一次get 的时候才被真正的初始化对象

使用配置：

druid.segmentCache.numBootstrapThreads=10
druid.segmentCache.lazyLoadOnStart=true           

收益：重启时间由 30Min 缩减到 3Min以内

本小结主要讲述了为了集群可用性 99.9% 的目标所做的工作，围绕着数据加载Pending问题，查询稳定性问题，慢启动造成的数据移动问题进行了原因定位分析，这些问题应该会伴随着集群规模和使用而出现，为了能够更好的提供服务，所要踩的坑。

Druid 使用建议

经过踩坑之旅，我们沉淀了对于离线数据导入和查询的使用建议

Druid 数据导入：

1. Druid 本身支持实时导入查询和批量导入两种模式，实时导入目前依赖 Tranquality 服务管理，批量导入通过 Hadoop MapReduce 导入 Hive 表数据。实时导入可支持实时查询，批量导入只有在导入完成时数据可查。
2. Druid 支持对导入数据按时间粒度预聚合，从而减少数据存储量，提高查询性能，因此要求在数据导入时必须设置预聚合时间粒度，粒度大小根据业务查询的最小时间精度决定。
3. Druid segment 的 shard 大小应该控制在 500 MB 以下，防止单个 shard 过大导致 OOM。
4. 针对批量导入，每次导入数据的 segment 不能过多，每次导入数据量不要超过 7 天

Druid 数据查询

1. 查询主要分为维度查询（topN/groupby）和非维度查询 (timeseries)
2. 维度查询中，避免 groupby 查询多个高基数维度（count(distinct(维度列))），会导致查询时内存爆炸，尽量不要导入高基数维度到 Druid 中，如果无法避免，只能用 topN 查询单个维度
3. 对于大时间范围的查询，可以拆分为多个小时间范围的查询，并行提交，以充分利用 Druid 多节点并行查询能力
4. 数据需要配置多副本，默认为 2 副本，以保障数据高可用，同时需要区分冷热，默认最近一个月为热数据（存放在 ssd 上），一个月之前的为冷数据 （存放在普通磁盘上），一年前的数据为冷备数据（存放在 HDFS 中，不可查询）

总结

首先本文主要围绕 Druid 集群在知乎的应用落地过程为主线，讲述了 Druid简介，在知乎的应用场景； 在落地的过程中经历了快速发展到稳定性痛点挑战的过程；最终经过平台架构V1.0到平台架构V2.0的架构迭代，解决了发展中遇到的痛点。

其次，本文中介绍的稳定性优化过程，平台有效的监控对于分析问题能够起到提效作用；问题明确后，积极参与社区讨论，这样能够提高问题解决的效率。

最后，希望读者可以收获到整个建设过程中遇到的问题和解决方案，对日常工作有所帮助！

作者:Jacky2020

出处:https://zhuanlan.zhihu.com/p/67607200