1-apache druid原理执行流程

Posted 2020-11-15 李昊宗

1、前言

　　从druid的0.11版本开始，我就开始关注它，每一次的版本的更新，druid都会使用户体验、性能更好，从以前手写配置文件到可视化的界面操作，从实时节点进行任务提交到现在的索引服务等

　　流处理：

　　　　　　日志监控（Flume/Airflow） ----> 消息中间件（kafka、MQ）  ----> 流处理（spark streaming/Flink)

　　　　　　为了使指标开发更加简单，我们一般不会直接采用Spark core或者Flink DataStream API进行计算，譬如很多时候都会通过样例类的方式，将其转化为table，然后利用SQL进行指标开发。

　　　　　　但是每一次的版本升级，我们如果要添加一些新的指标，那么只能更改原始代码或者重新跑一个任务

　　　　　　并且有的时候，数据分析师往往需要的是自主查询，自己定义指标、自己去开发指标！！

　　由于以上的困境，druid帮我们解决了很大一部分，当然它并非完美，在后面我会提到！

　　在此之前，我先抛出两个概念：

　　　　OLTP：

　　　　　　　　　　对数据的增删改查等操作，主要用于传统的关系数据库。

　　　　OLAP：

　　　　　　　　　　数据按不同维度的聚合，维度的下钻，上卷等，主要用于数据仓库。

2、druid

　　　　　　概念：
　　　　　　　　　　主要是解决低延迟下实时数据摄入与查询的平台，本质是一个数据存储，但是数据仍然是保存在（hdfs、文件系统等）中。

　　　　　　特点：

　　　　　　　　　　①列式存储格式：

　　　　　　　　　　　　可以将列作为索引，为仅查看几列的查询提供了巨大的速度提升

　　　　　　　　　　②高可用、高并发：

　　　　　　　　　　①集群扩展、缩小、删除、宕机都不会停止服务，全天候运行
　　　　　　　　　　②HA、sql的并行化执行、可扩展、容灾等
　　　　　　　　　　③支持1000+的并发用户，并提供隔离机制支持多租户模式（多租户就是并发互不影响）

　　　　　　　　　　④低延迟

　　　　　　　　　　　　　Druid采用了列式存储、倒排索引、位图索引等关键技术，能够在亚秒级别内完成海量数据的过滤、聚合以及多维分析等操作。

　　　　　　　　　　⑤存储时候聚合：

　　　　　　　　　　　　　无论是实时数据消费还是批量数据处理，Druid在基于DataSource结构存储数据时即可选择对任意的指标列进行聚合操作

　　　　　　　　　　　　　聚合：提前做好sum，count等操作

3、druid架构

　　　　四个节点 +  三个依赖

　　　　四个节点：

　　　　　　　　实时节点（Realtime Node）：

　　　　　　　　　　　　实时摄入数据，对于旧的数据周期性的生成segment数据文件，上传到deep storage中

　　　　　　　　　　　　为了避免单点故障，索引服务(Indexer)的主从架构已经逐渐替代了实时节点:

　　　　　　　　　　　　　　　可以通过索引服务的API，写数据导入任务，用以新增、删除、合并Segment等。是一个主从架构：

　　　　　　　　　　　　　　统治节点（overlord）：

　　　　　　　　　　　　　　　　　　　　　　　　　　　　 　类似于Yarn ResourceManager ： 负责集群资源的管理和分配

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　监视数据服务器上的MiddleManager进程，将提取任务分配给MiddleManager

　　　　　　　　　　　　　　中间管理者（middle manager）：

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　类似于Yarn NodeManager ： 负责单个节点资源的管理和分配

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　新数据提取到群集中的过程。他们负责从外部数据源读取并发布新的段

　　　　　　　　　　　　　　苦工（peon）：

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　 类似于Yarn container ：负责具体任务的执行

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　Peon进程是由MiddleManagers产生的任务执行引擎。
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　每个Peon运行一个单独的JVM，并负责执行单个任务。

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　Peon总是与生成它们的MiddleManager在同一主机上运行

　　　　　　　　　　　　　　Router
　　　　　　　　　　　　       (路由：可选)：

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　可在Druid代理，统治节点和协调器之前提供统一的API网关

　　　　　　　　　　　　　注：统治节点和中间管理者的通信是通过zookeeper完成的

　　　　　　　　　历史节点（Historical Node）：

　　　　　　　　　　　　　加载已生成的segment数据文件，以供数据查询

　　　　　　　　　　　　　启动或者受到协调节点通知的时候，通过druid_rules表去查找需要加载的数据，然后检查自身的本地缓存中已存在的Segment数据文件，
　　　　　　　　　　　　　然后从DeepStorage中下载其他不在本地的Segment数据文件，后加载到内存！！！再提供查询。

　　　　　　　　　查询节点（Broker Node）:

　　　　　　　　　　　　　对外提供数据查询服务，并同时从实时节点与历史节点查询数据，合并后返回给调用方

　　　　　　　　　　　　   缓存：
　　　　　　　　　　　　　　外部：第三方的一些缓存系统
　　　　　　　　　　　　　　内部：在历史节点或者查询节点做缓存

　　　　　　　　　协调节点（Coodinator Node）:

　　　　　　　　　　　　负责历史节点的数据负载均衡，以及通过规则（Rule）管理数据的生命周期

　　　　　　　　　　　　①通过从mysql读取元数据信息，来决定深度存储上哪些数据段应该在那个历史节点中被加载，
　　　　　　　　　　　　②通过ZK感知历史节点，历史节点增加，会自动分配相关的Segment，历史节点删除，会将原本在这台节点上的Segment分配给其他的历史节点

　　　　　　　　　　　　注：Coordinator是定期运行的，并且运行间隔可以通过配置参数配置

　　　　三个依赖：

　　　　　　　　1） Mysql：

　　　　　　　　　　　　存储关于Druid中的metadata，规则数据，配置数据等，
　　　　　　　　　　　　主要包含以下几张表：
　　　　　　　　　　　　　　"druid_config”（通常是空的）,
　　　　　　　　　　　　　　“druid_rules”（协作节点使用的一些规则信息，比如哪个segment从哪个node去load）
　　　　　　　　　　　　　　“druid_segments”（存储 每个segment的metadata信息）;

　　　　　　　　2 ）Deep storage：

　　　　　　　　　　　　存储segments，Druid目前已经支持本地磁盘，NFS挂载磁盘，HDFS，S3等。

　　　　　　　　3） ZooKeeper：

　　　　　　　　　　　　①查询节点通过Zk来感知实时节点和历史节点的存在，提供查询服务。
　　　　　　　　　　　　②协调节点通过ZK感知历史节点，实现负载均衡
　　　　　　　　　　　　③统治节点、协调节点的lead选举

4、DataSource

　　　　　　Druid中的数据存储在被称为datasource中，类似RDMS中的table！！！

　　　　　　每个datasource按照时间划分。每个时间范围称为一个chunk(一般都是以天分区，则一个chunk为一天)！！！ //也可以按其他属性划分

　　　　　　在chunk中数据被分为一个或多个segment，每个segment都是一个单独的文件，通常包含几百万行数据

　　　　　　注：这些segment是按照时间组织成的chunk，所以在按照时间查询数据时，效率非常高。

 

　　　　　　数据分区：

　　　　　　　　任何分布式存储/计算系统，都需要对数据进行合理的分区，从而实现存储和计算的均衡，以及数据并行化。

　　　　　　　　而Druid本身处理的是事件数据，每条数据都会带有一个时间戳，所以很自然的就可以使用时间进行分区。

　　　　　　　　为什么一个chunk中的数据包含多个segment！！！？？？？原因就是二级分区

　　　　　　二级分区：

　　　　　　　　很可能每个chunk的数据量是不均衡的，而Duid为了解决这种问题，提供了“二级分区”，每一个二级分区称为一个Shard(分片)
　　　　　　　　其实chunk、datasource都是抽象的，实际的就是每个分区就是一个Shard，每个Shard只包含一个Segment！！！，因为Segment是Shard持久化的结果

　　　　　　　　Druid目前支持两种Shard策略：

　　　　　　　　Hash(基于维值的Hash)
　　　　　　　　Range(基于某个维度的取值范围)

　　　　　　　　譬如：

　　　　　　　　　　2000-01-01，2000-01-02中的每一个分区都是一个Shard
　　　　　　　　　　2000-01-02的数据量比较多，所以有两个Shard。每个分区都是一个Shard

　　　　　　　　　　Shard经过持久化之后就称为了Segment，Segment是数据存储、复制、均衡(Historical的负载均衡)和计算的基本单元了。
　　　　　　　　　　Segment具有不可变性，一个Segment一旦创建完成后(MiddleManager节点发布后)就无法被修改，
　　　　　　　　　　只能通过生成一个新的Segment来代替旧版本的Segment。　　

　　　　　　　Segment内部存储结构：

　　　　　　　　　　Segment内部采用列式存储 //并不是说每列都是一个独立的文件，而是说每列有独立的数据结构，所有列都会存储在一个文件中

　　　　　　　　　　Segment中的数据类型主要分为三种：

　　　　　　　　　　　　时间戳
　　　　　　　　　　　　维度列
　　　　　　　　　　　　指标列

　　　　　　　　　　对于时间戳列和指标列，实际存储是一个数组

　　　　　　　　　　对于维度列不会像指标列和时间戳这么简单，因为它需要支持filter和group by：

　　　　　　　　　　所以Druid使用了字典编码(Dictionary Encoding)和位图索引(Bitmap Index)来存储每个维度列。每个维度列需要三个数据结构：

　　　　　　　　　　　　　　1、需要一个字典数据结构，将维值(维度列值都会被认为是字符串类型)映射成一个整数ID。
　　　　　　　　　　　　　　2、使用上面的字典编码，将该列所有维值放在一个列表中。
　　　　　　　　　　　　　　3、对于列中不同的值，使用bitmap数据结构标识哪些行包含这些值。　　　　　　 //位图索引，这个需要记住

　　　　　　　　　　　　　　注：使用Bitmap位图索引可以执行快速过滤操作(找到符合条件的行号，以减少读取的数据量)

　　　　　　　　　　Druid针对维度列之所以使用这三个数据结构，是因为：

　　　　　　　　　　　　　　使用字典将字符串映射成整数ID，可以紧凑的表示结构2和结构3中的值。
　　　　　　　　　　　　　　使用Bitmap位图索引可以执行快速过滤操作(找到符合条件的行号，以减少读取的数据量)，因为Bitmap可以快速执行AND和OR操作。
　　　　　　　　　　　　　　对于group by和TopN操作需要使用结构2中的列值列表

　　　　　　　　　　　　　　实例：

　　　　　　　　　　　　　　　　1. 使用字典将列值映射为整数
　　　　　　　　　　　　　　　　　　{
　　　　　　　　　　　　　　　　　　　　"Justin Bieher":0,
　　　　　　　　　　　　　　　　　　　　"ke$ha":1
　　　　　　　　　　　　　　　　　　}
　　　　　　　　　　　　　　　　2. 使用1中的编码，将列值放到一个列表中

　　　　　　　　　　　　　　　　　　[0,0,1,1]

　　　　　　　　　　　　　　　　3. 使用bitmap来标识不同列值

　　　　　　　　　　　　　　　　　　value = 0: [1,1,0,0] //1代表该行含有该值，0标识不含有
　　　　　　　　　　　　　　　　　　value = 1: [0,0,1,1]

　　　　　　　　　　　　　　　　　　因为是一个稀疏矩阵，所以比较好压缩！！
　　　　　　　　　　　　　　　　　　Druid而且运用了Roaring Bitmap能够对压缩后的位图直接进行布尔运算，可以大大提高查询效率和存储效率(不需要解压缩)

　　　　　　　Segment命名：

　　　　　　　　　　　　　　如果一个Datasource下有几百万个Segment文件，我们又如何快速找出我们所需要的文件呢？答案就是通过文件名称快速索引查找。

　　　　　　　　　　　　　　Segment的命名包含四部分：

　　　　　　　　　　　　　　　　　　数据源(Datasource)、时间间隔(包含开始时间和结束时间两部分)、版本号和分区(Segment有分片的情况下才会有)。

　　　　　　　　　　　　　　　　　　eg：wikipedia_2015-09-12T00:00:00.000Z_2015-09-13T00:00:00.000Z_2019-09-09T10:06:02.498Z

　　　　　　　　　　　　　　　　　　　　wikipedia： Datasource名称
　　　　　　　　　　　　　　　　　　　　开始时间： 2015-09-12T00:00:00.000Z      //该Segment所存储最早的数据，时间格式是ISO 8601
　　　　　　　　　　　　　　　　　　　　结束时间： 2015-09-13T00:00:00.000Z      //该segment所存储最晚的数据，时间格式是ISO 8601
　　　　　　　　　　　　　　　　　　　　版本号： 2019-09-09T10:06:02.498Z         //此Segment的启动时间，因为Druid支持批量覆盖操作，
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　        //当批量摄入与之前相同数据源、相同时间间隔数据时，数据就会被覆盖，这时候版本号就会被更新
　　　　　　　　　　　　　　　　　　      分片号： 从0开始，如果分区号为0，可以省略 //分区的表现其实就是分目录

　　　　　　　　　　　　　　注：单机形式运行Druid，这样Druid生成的Segment文件都在${DRUID_HOME}/var/druid/segments 目录下
　　　　　　　　　　　　　　注：为了保证Druid的查询效率，每个Segment文件的大小建议在300MB~700MB之间
　　　　　　　　　　　　　　注：版本号的意义：

　　　　　　　　　　　　　　　　　　　　在druid，如果您所做的只是追加数据，那么每个时间chunk只会有一个版本。

　　　　　　　　　　　　　　　　　　　　但是当您覆盖数据时，因为druid通过首先加载新数据（但不允许查询）来处理这个问题，一旦新数据全部加载，
　　　　　　　　　　　　　　　　　　　　切换所有新查询以使用这些新数据。然后它在几分钟后掉落旧段！！！

　　　　　　　　存储聚合：

　　　　　　　　　　　　　　无论是实时数据消费还是批量数据处理，Druid在基于DataSource机构存储数据时即可选择对任意的指标列进行聚合操作：

　　　　　　　　　　　　　　1、基于维度列：相同的维度列数据会进行聚合
　　　　　　　　　　　　　　2、基于时间段：某一时间段的所有行会进行聚合，时间段可以通过queryGranularity参数指定

　　　　　　　　　　　　　　聚合：提前做好sum，count等操作

　　　　　　　　Segment声明周期：

　　　　　　　　　　　　　　在元数据存储中！每个Segment都会有一个used字段，标记该段是否能用于查询

　　　　　　　　　　　　　　is_Published：

　　　　　　　　　　　　　　　　　　当Segment构建完毕，就将元数据存储在元数据存储区中，此Segment为发布状态

　　　　　　　　　　　　　　is_available:

　　　　　　　　　　　　　　　　　　如果Segment当前可用于查询（实时任务或历史进程），则为true。

　　　　　　　　　　　　　　is_realtime:

　　　　　　　　　　　　　　　　　　如果是由实时任务产生的，那么会为true，但是一段时间之后，也会变为false

　　　　　　　　　　　　　　is_overshadowed：

　　　　　　　　　　　　　　　　　　标记该段是否已被其他段覆盖！处于此状态的段很快就会将其used标志自动设置为false。

 

5、实时Segment数据文件的流动：

　　　　　　生成:

　　　　　　　　①实时节点（中间管理者）会周期性的将同一时间段生成的数据合并成一个Segment数据文件，并上传到DeepStorage中。　　　　　　　　

　　　　　　　　②Segment数据文件的相关元数据信息保存到MetaStore中（如mysql，derby等）。　　　　　　　　

　　　　　　　　③协调节点定时（默认1分钟）从MetaSotre中获取到Segment数据文件的相关元信息后，将按配置的规则分配到符合条件的历史节点中。　　　　　　　　

　　　　　　　　④协调节点会通知一个历史节点去读　　　　　　　　

　　　　　　　　⑤历史节点收到协调节点的通知后，会从DeepStorage中拉取该Segment数据文件到本地磁盘，并通过zookeeper向集群声明可以提供查询了。

　　　　　　　　⑥实时节点会丢弃该Segment数据文件，并通过zookeeper向集群声明不在提供该Sgment的查询服务。 //但是第四步已经可以提供查询服务了
　　　　　　　　⑦而对于全局数据来说，查询节点（Broker Node）会同时从实时节点与历史节点分别查询，对结果整合后返回用户。

　　　　　　查询：

　　　　　　　　查询首先进入Broker，按照时间进行查询划分
　　　　　　　　确定哪些历史记录和 MiddleManager正在为这些段提供服务
　　　　　　　　Historical / MiddleManager进程将接受查询，对其进行处理并返回结果

这篇博客虽然很枯燥，但是下一篇博客会用这个给大家做一个实例，让大家更好的了解！！！