Druid数据结构及架构原理

一、Druid数据结构

Druid中的数据存储在datasource中，类似RDBMS中的table，每个datasource中按照时间划分，每个时间范围称为一个chunk(一般一天为一个chunk)，在一个chunk中数据根据维度的Hash或者范围被分成一个或者多个segment，每个segment都是一个单独的文件，通常包含几百万行数据，这些segment是按照时间组织成的，所以在按照时间查询数据时，效率非常高。实际上，datasource和chunk都是抽象的，Druid底层存储就是Segment，一旦一个Segment生成后就无法被修改，只能通过生成一个新的Segment来代替旧版本的Segment。

Segment内部采用列式存储，并不是每个列都对应一个独立的文件，而是每列都有独立的数据结构，这些列存储在一个文件中。

在Segment中数据类型有三种：时间戳、维度列、指标列，举例如下：

对于时间戳和指标列，实际存储是一个数组。对于维度列而言，由于需要支持filter和group by操作，所以Druid使用了字典编码（Dictionary Encoding）和位图索引（Bitmap Index）来存储每个维度列。每个维度列需要三个数据结构：

• 需要一个字典数据结构，将维度值映射成一个整数ID
• 使用上面的字典编码，将该列所有维值放在一个列表中。
• 对于列中不同的值，使用bitmap数据结构标识哪些行包含这些值。

例如，有如下表数据：

Druid将以上数据存储构建的位图索引如下，来实现数据快速查找：

1. 将维度列中的值映射成一个个整数ID

维度列有loc，item两列，其中的不重复值有：北京、上海、广州、杭州、书籍、电脑、手机、水果，假设映射后的整数ID组成的字典编码如下：

北京-0，上海-1，广州-2，杭州-3，书籍-4，电脑-5，手机-6，水果-7

1. 构建的位图索引如下

我们要过滤找到“北京”并且“电脑”的数据，那么只需要在位图索引中找到北京对应的行和广州对应的行做与运算即可。假设根据上表数据我们查询sql如下：

select sum(amount) as totalamount 
from tbl 
where loc = “北京”and “item”=“电脑”

那么根据以上位图，我们可以看到条件“loc=北京”对应的bitmap位图为(110001),”item=电脑”对应的位图为(011000)，查询条件就是（110001） and(011000)结果就是找(010000)所对应的行，也就是找出第2行数据即可。

同理，如果要按照“loc”和”“item”分组，只需要找到“loc”下各个值与“item”下各个值进行与运算即可。

Segment命名可以让我们在大量的Segment文件中快速找到我们查询的Segment文件，Segment命名包含四个部分：数据源(DataSource)、时间间隔（包含开始时间和结束时间两部分）、版本号和分区(当一个Chunk中有多个Segment时会有分区号)。

例如：

kafkadata_2022-07-30T00:00:00.000Z_2022-07-31T00:00:00.000Z_2022-08-02T12:02:03.456Z_1

• kafkadata : DataSource名称。
• 2022-07-30T00:00:00.000Z：开始时间，该Segment存储最早的数据。
• 2022-07-31T00:00:00.000Z：结束时间，该Segment存储最晚的数据。
• 2022-08-02T12:02:03.456Z：版本号，该Segment启动时间，由于Druid支持批量覆盖操作，当批量摄入与之前相同数据源，相同时间间隔数据时，数据就会被覆盖，这时候版本号就会被更新。在Druid中，如果只是追加数据，那么每个时间chunk只有一个版本，如果覆盖数据时，Druid会加载全部新数据生成一个新的版本号，然后会删除旧的Segment。
• 1：分区号，如果分区号为0，会省略，分区的表现其实就是分目录。

注意：单机运行Druid，Druid生成的Segment都在$DRUID_HOME/var/druid/segment目录下。

二、Druid架构原理

下图是Druid的架构图，蓝色代表Druid服务，黄色代表Druid的进程，每个进程所在的节点我们也可以称为当前进程节点，例如：Borker进程所在节点，可以称为该节点为Broker节点。整体来看Druid可以分为四类节点和三个依赖。

1、四类节点

1.1、实时节点（RealTime Node）

实时类节点包含很多进程角色，也可以称为很多，为了防止单点故障，所以这些角色没有在一台节点启动，实时类节点包含的进程如下：

• Overlord（统治进程）

该进程监视MiddleManager进程，并且是数据摄入Druid的控制器，负责将提取任务分配给MiddleManagers并协调Segement发布。类似Yarn中的ResourceManager,负责集群资源的管理和分配。

• MiddleManager（中间管理进程）

类似Yarn中NodeManager,负责单个节点的资源管理和分配。该进程负责接收实时数据和批数据，生成Segment数据文件。

• Router

该进程可以将请求路由到Broker、Coordinator、Overlord。

1.2、协调节点（Coodinator Node）

• Coordinator（协调进程）

主要负责Historical的数据负载均衡，以及通过规则（Rule）管理数据的生命周期。默认每隔1分钟同步元数据库，感知新生成的Segment，将待加载的Segment信息保存在Zookeeper目录中（此目录是对应在线的Historical Node在Zookeeper中的目录），Historical Node感知到需要加载新的Segment时，首先去本地磁盘目录下检查该Segment是否已下载，如果没有，会从zookeeper中获取对应的元数据（该Segment存储在何处，如果解压等）获取该文件并加载到内存中，同时在zookeeper对应目录下标记声明该Segment已被加载，从而该Segment可以被查询。

协调进程除了告诉Historical加载新数据之外，还会负责协调卸载过期数据、复制数据、和为了负载均衡移动数据。

如果集群内所有的Coordinator Node都停止服务，整个集群对外依然有效，不过新Segment不会被加载，过期的Segment也不会被丢弃，即整个集群内的数据拓扑会一直保持不变，直到新的Coordinator Node服务上线。

1.3、历史节点（Historical Node）

• Historical（历史进程）

该进程负责存储已生成好的Segment数据文件，以供数据查询。当收到Coordinator进程通知的时候，检查本地缓存中已经存在的Segment数据文件，然后从DeepStorage中下载其他不在本地的Segment数据文件加载到内存提供查询。Historical进程是整个集群查询性能的核心所在，承担大部分的Segment查询。

1.4、查询节点（Broker Node）

• Borker（查询进程）

该进程负责客户端数据的查询，并将这些查询转发给Historical和middleManager。

Druid进程可以以任意方式进行部署，为了方便部署，建议分为三种服务器类型：主服务器（Master）、查询服务器（Query）、数据服务器（Data）。

• Master:运行Coordinator和Overlord进程，负责集群可用和读取数据。
• Query:运行Broker和Router进程，负责处理外部客户端的查询。
• Data:运行Historical和Middle Manager进程，负责数据接收和所有可查询数据的存储。

2、​​​​​​​三个外部依赖

2.1、Metadata Storage

元数据存储，存储Druid集群的元数据信息，比如：Segment的相关信息，一般用MySQL或者PostgreSQL。

2.2、Zookeeper

为Druid集群提供协调服务，例如：

• Broker Node查询节点通过Zookeeper来感知实时节点和历史节点的存在，提供查询服务。
• 协调节点通过zookeeper感知历史节点，实现负载均衡。
• Overlords 统治节点、Coordinators协调节点的Leader的选举。

2.3、Deep Storage

数据文件存储，存放生成的Segment数据文件，并供Historical节点进行下载，对于单节点集群可以使本地磁盘，对于分布式集群一般是HDFS，也可以是NFS挂载磁盘、S3等。

3、​​​​​​​​​​​​​​数据写入及读取原理

上图中描述了Druid集群内部实时/批量数据写入以及查询的请求过程。当实时数据达到Realtime Node后，在时间窗口内的数据会停留在RealTime Node内存中，而时间窗口外的数据会组织成Segment存储到Deep Storage中，同时Segment的元数据信息会被注册到元数据库中（这里是MySQL）,Coordinator Nodes会定期（默认1分钟）去同步原信息库，感知新生成的Segment，并通知Historical Nodes去加载Segment,Zookeeper也会更新整个集群内部数据元数据信息。

当用户在客户端查询信息时，会将请求提交给Broker Node,Broker Node会请求Zookeeper获取集群内数据分布信息，从而获取将请求发送给哪些Historical Node以及Realtime Node，汇总各节点的返回数据并将最终结果返回给用户。