Druid概念
Druid.io 是一个开源的，分布式的，列式存储的，适用于实时数据分析的OLAP系统，能够快速聚合、灵活过滤、毫秒级查询、和低延迟数据导入。
优势：
- 高容错性：单个节点挂掉不会影响其他部分。详情在文末有介绍。
- 多版本控制（MVCC）：通过数据更新时间来区分版本，历史节点只加载最新版本数据。支持实时数据indexing与批量数据indexing同时进行，实时数据索引满足实时需求，批量数据覆盖实时数据满足准确性需求。详情见druid数据加载。
- 高聚合性：数据roll up预处理，减少存储数据量。详情见druid预聚合roll-up。
- 高压缩度：使用Bitmap indexing加速列存储的查询速度，并使用CONCISE算法来对bitmap indexing进行压缩，使得生成的segments比原始文本文件小很多
集群组成
Druid集群包含不同类型的节点，而每种节点都被设计来做好某组事情。这样的设计可以隔离关注并简化整个系统的复杂度。不同节点的运转几乎都是独立的并且和其他的节点有着最小化的交互，因此集群内的通信故障对于数据可用性的影响非常小。
主要分为四大部分：数据生产、数据存储、数据查询、外部依赖。本小节先做一个简单的介绍，后面每个部分会有详细介绍。
数据生产
- Indexing Service 索引服务节点：
由多个worker组成的集群，负责为加载批量的和实时的数据创建索引，并且允许对已经存在的数据进行修改。 - Realtime 实时节点：
负责加载实时的数据到系统中，在生产使用的几个限制成本上实时节点比索引服务节点更容易搭建。
实时数据和批量数据加载的两种方式在上一篇druid数据相关中介绍到了，就不赘述。
数据存储
- Coordinator 协调节点：
对历史节点的分组进行监控，以确保数据可用，和最佳的配置。协调节点通过从元数据存储中读取元数据信息来判断哪些segments是应该加载到集群的，使用Zookeeper去判断哪些历史节点是存活的，在Zookeeper中创建任务条目告诉历史节点去加载和删除segments。 - Historical 历史节点：
负责处理历史数据存储和查询历史数据（非实时），历史节点从“deep storage”下载segments，将结果数据返回给broker节点，historical加载完segment通知Zookeeper，Historical nodes使用Zookeeper监控需要加载或者删除哪些新的segments。
数据查询
- Broker 代理节点
接收来自外部client的查询请求，并转发这些请求给实时节点和历史节点，当代理节点接收到结果时，将来自实时节点和历史节点的结果合并返回给调用方。为了知道整个拓扑结构，代理节点通过使用Zookeeper在确定哪些实时节点和历史节点存活。
外部依赖
Druid的集群需要有一些外部依赖。
- Zookeeper
- Metadata Storage
- Deep Storage
Indexing Service
索引服务负责实时和批量数据的导入、分析、索引、压缩，生成“segment”（数据段），存入Deep Storage（例如HDFS）。
实时数据通过Tranquility客户端，批量数据通过Batch Data Ingestion，将任务提交给overload，分配给Middle Manager创建Poen执行。
结构
索引服务是主从结构，由三个部分组成：
- peon组件：在一个单独的jvm中运行单个任务，通过单独的jvm对任务做资源隔离和日志隔离。
- Middle Manager：用于创建和管理peon的中层管理组件
- overlord组件：管理任务分配到Middle Manager
综合Tranquility和整个系统之后：
流程：
- 用户的spec文件在Tranquility中定义，首先Tranquility通过spec初始化，获得zk中Overlord的地址，与Overlord通信。
- Overlord得到新写入任务后，查询zk节点信息，选择一个Middle Manager节点启动来启动peon，并将信息写入到zk中。
- Middle Manager一直监控zk，发现有新的任务分配后，启动一个Peon进程，并监控Peon进程的状态。
- Peon与Realtime Node流程基本一致，所不同的是Peon使用的是HTTP接口来接收数据，RealTime Node更多的是内部的线程不断的拉取Kafka的数据。
- Tranquility随后通过zk获取Peon机器地址和端口，将数据不断的发送到Peon中。
- Peon根据spec规则，定时或者定量将数据build index，handoff到deep storage(HDFS)中。
- Coordinator根据Peon在zk中信息，将元数据写入到mysql中，并分配Historical Node去deep storage拉取index数据。
- Historical Node到deep storage拉取index数据到本地，重建index到内存中，至此数据流入完成。
配置说明
overload节点配置时
druid.indexer.runner.type=local表示overload以本地模式运行，overload同时负责Middle Manager的作用，创建poen和分配任务。local模式下，overload配置需要对poen进行配置。
real-time node
实时节点是进行存储和查询实时数据的工作区。在Zookeeper中通告它们的在线状态和为哪些数据提供服务。
存储：metadata(元数据)写入MySQL，在ZooKeeper中新增一条记录
Segment定期会转存到DeepStorage
查询：提供实时查询索引，响应broker的查询
下图中的master即为coordinator：
具体存储过程：
- 实时节点缓存事件数据到内存中的索引上，然后有规律的持久化到磁盘上。在转移之前，持久化的索引会周期性地合并在一起。（查询会同时命中内存中的和已持久化的索引。）
- 实时节点周期性的启动后台的计划任务搜索本地的持久化索引，后台计划任务将这些持久化的索引合并到一起并生成一块不可变的数据，这些数据块包含了
- 一段时间内的所有已经由实时节点导入的事件数据，称这些数据块为”Segment”。
- 在传送阶段，实时节点将这些segment上传到一个永久持久化的备份存储中,即Deep Storage
realtime node与indexing service导入数据的区别
| 比较项 | RealTime Node | Realtime Indexing Service |
|---|---|---|
| 角色 | RealTime Node | Overloard Nodes，Middle Manager Nodes，Poens |
| 部署方式 | 多台服务器或单台服务器上多个RTN，每个RTN指定不同的spec文件启动 | 仅部署Overloard Nodes和Middle Manager。Middle Manager创建poen去接收不同的realtime日志，不需要指定spec文件，由Tranquility客户端提供 |
| 使用方式 | RTN通过spec文件指定消费的Kafka topic，不断的pull | Poen接收Tranquility客户端push过来的数据 |
| 可扩展性与易用性 | 通过加机器，启动更多的RealTime Nodes进行扩展，但是需要管理所有的RTN的spec文件，消费的各种属性信息，运维复杂 | 通过加机器，启动更多的Middle Managers来进行扩展，所有的日志消费属性信息都是通过Tranquility自己指定，运维简单 |
随着Druid业务增多，规模扩大，对Realtime Node的管理变成了非常繁琐的事情，使用Realtime Index Service是必须的
Coordinator Node
协调节点可以认为是Druid中的master，通过Zookeeper管理历史节的segment放置策略，且通过Mysql中的metadata管理数据段。主要作用：
- 通过从元数据存储（MySQL）读取数据段的元数据信息，来决定哪些Segments应该在集群中被加载。
- 使用ZK来确定哪个Historical节点存在
- 创建ZK条目告诉Historical节点管理Segments：加载新的Segments,删除旧的Segments,或者移动Segments进行负载均衡。
负载均衡
协调节点每次运行时，
计算每一个historical tier（节点层）利用率最高的和最低的差异值。
如果差异超过某个阈值,部分segment将从最高的节点迁移到最低的节点，并且移动的segment是随机选择的。
每次协调节点运行时，能够迁移的segment的数量是可配置的。
例子：
在实际生产过程中，hot节点层本身只有两台机器，增加hot层到5台。负载最高的机器数据会往低的迁移。到最后每台机器上的数据大致相等。
Historical Node
历史节点负责加载历史Segment并且提供针对这些历史Segment的查询。Historical Nodes可分为多个tier， 比如热数据放在一个tier， 冷数据放到另外一个tier，以达到冷热数据分开处理的目的。
Historical节点segment创建流程
- Coordinator在ZK下与Historical节点相关联的加载队列路径下创建一个临时记录。
- 每个历史节点与ZK保持一个长连接监测ZK。
- 当一个历史节点发现在Zookeeper中与它关联的加载队列目录下有一个新的加载记录时。
- 它首先检查本地磁盘目录（缓存）中关于新的Segment的信息。如果缓存中没有关于新的Segment的信息，历史节点将下载新的Segment的元数据信息并告知Zookeeper。元数据包含新的Segment在“Deep Storage”中的存储位置，怎样去解压缩和处理新的Segment的信息。
- 一旦一个历史节点处理完成一个Segment，该Segment在Zoookeeper与该节点关联的服务Segments路径中公布可以提供服务。
- 此刻，这个Segment可以用于查询。
Broker Node
broker（代理）提供针对segment的路由查询，代理节点从Zookeeper获取Segments存储在哪些节点和怎样找到正确的节点的元数据。将查询转发到Realtime和Historical节点。把来自于所有单个节点的结果合并在一起。
转发查询
Zookeeper维护有关历史和实时的节点信息和他们所能提供服务的Segment。
在Zookeeper的每一个数据源，代理节点建立相关Segments的时间轴和为这些Segments提供服务的节点。
当收到一个特定数据源和时间间隔的查询请求，代理节点执行查找与查询数据源时间间隔相关的时间轴和检索包含数据查询的节点。代理节点然后将查询转发到所选节点。
缓存策略
Broker节点包含一个支持LRU失效策略的缓存。
- 首先将这个查询映射到一组Segments，这些Segment结果的子集可能在缓存中已经存在，在缓存中已经存在的结果可以被直接拉取。
- 对于一些缓存中不存在的结果。代理节点会转发查询到历史节点。一旦历史节点返回其结果，代理节点将结果存储到缓存中。
- 实时节点返回的结果将永远不会被缓存，因此实时节点的查询请求将永远被转发到实时节点。实时节点的数据是不断变化的，缓存实时节点的结果是不可靠的。
外部依赖
Metadata Storage（Mysql）
存储segments的元数据和配置，而不是存储实际数据，包含3张表：
“druid_config”（通常是空的）,
“druid_rules”（协作节点使用的一些规则信息，比如哪个segment从哪个node去load）
“druid_segments”（存储 每个segment的metadata信息）
Deep storage
segments的永久备份，Druid目前已经支持本地磁盘、NFS挂载磁盘、HDFS、S3等。
创建segments的服务上传segments到Deep storage,然后historical节点下载。
ZooKeeper
管理当前集群状态, 发现和维持当前的数据拓扑。（节点状态、数据操作、数据同步）
- realtime-node和historical-node在Zookeeper中通告它们的在线状态和为哪些数据提供服务。
- 管理当前cluster的状态，比如记录哪些segments从实时节点移到了历史节点
主要发生：
- 协调节点的leader选举
- 历史和实时节点发布segment协议
- 协调节点和历史节点之间的segment Load/Drop协议
- overlord的leader选举
- 索引服务任务管理
数据流过程
- 实时数据写入到实时节点,会创建索引结构的Segment
- 实时节点的Segment经过一段时间会转存到DeepStorage
- 元数据写入MySQL; 实时节点转存的Segment会在ZooKeeper中新增一条记录
- 协调节点从MySQL获取元数据,比如schema信息(维度列和指标列)
- 协调节点监测ZK中有新分配/要删除的Segment,写入ZooKeeper信息:历史节点需要加载/删除Segment
- 历史节点监测ZK, 从ZooKeeper中得到要执行任务的Segment
- 历史节点从DeepStorage下载Segment并加载到内存/或者将已经保存的Segment删除掉
- 历史节点的Segment可以用于Broker的查询路由
说明
上述流程中的实时节点，换成indexing service，流程基本一致。
容错性
历史节点挂掉
该节点就不会服务这个节点上的Segments。
但是只要这些Segments仍然存在于DeepStorage,其他节点就会下载它们并服务这些Segments。
可以从集群中移除所有的历史节点,并且重新发布它们,也不会有任何的数据损失(因为数据最终都保存到DeepStorage中)
DeepStorage不可用
历史节点上已经加载了DeepStorage的Segments,仍然可用于查询。
但是新进来的数据无法进入到集群中(DS挂掉)。
协调节点挂掉
数据的拓扑(data topology)停止服务,就不会有新的数据以及数据的负载均衡。因为协调节点会通知历史节点下载新数据。
如果实时节点将Segment转存到DeepStorage,而没有历史节点去下载这些数据,会导致实时节点最终会丢弃这份过期的数据。
Broker挂掉
还会有其他的Broker接管请求,但是要至少保证有多余的Broker。
当然如果不向Broker发送请求,而只关心最新的实时数据,可以直接访问实时节点. 不过这种情况是很少见的.
实时节点
根据发送流的语义,可以有多个实时节点或者tranquility同时运行,处理同一个输入流,每个实时节点分担输入流的一部分数据.
元数据存储挂掉
协调节点就无法找到集群中新的Segments(因为新的Segment一定会写入记录到元数据存储中)。但仍然可以提供当前集群的数据视图.
ZooKeeper挂掉
数据拓扑不会被更新(同协调节点挂掉),但是Broker仍然可以维护最近的数据拓扑,并继续提供查询的服务。
参考
druid.io
druidio.cn
lxw的大数据田地–Druid.io实时OLAP数据分析存储系统介绍
萌の宇博客–realtime node与index server区别
zqhxuyuan博客–Druid OLAP架构设计