Druid - 概览

1.Druid是什么

     Druid是一个分布式的支持实时分析的数据存储系统，遵循的三个设计：

• 快速查询：通过数据预聚合+数据内存化+索引实现快速查询

• 水平扩展能力：分布式数据存储+并行化查询

• 实时分析：不可变的过去，只追加未来

 

     Druid具有如下的技术特点：

• 数据吞吐量大

• 支持流式数据的摄入和实时分析

• 查询灵活快速

• 社区支持力度大

 

    Druid数据格式：

• 时间列：以时间序列进行数据分片，所有查询以时间为中心。

• 维度列：Druid基于列式存储，查询结果展示列，常用于数据过滤，如示例数据集有四个维度:出版商，广告商，性别和国家。

• 指标列：通常用于计算值，操作方法如：COUNT、SUM等。

timestamp             publisher          advertiser  gender  country  click  price
2011-01-01T01:01:35Z  bieberfever.com    google.com  Male    USA      0      0.65
2011-01-01T01:03:63Z  bieberfever.com    google.com  Male    USA      0      0.62
2011-01-01T01:04:51Z  bieberfever.com    google.com  Male    USA      1      0.45
2011-01-01T01:00:00Z  ultratrimfast.com  google.com  Female  UK       0      0.87
2011-01-01T02:00:00Z  ultratrimfast.com  google.com  Female  UK       0      0.99
2011-01-01T02:00:00Z  ultratrimfast.com  google.com  Female  UK       1      1.53

    Druid数据分片：

• Druid的分片称之为Segment段，通常按时间对数据进行分片。如对示例数据进行压缩，我们可以创建两个段，按每小时分片。

• 段是保存时间间隔内数据，段包含按列存储的数据以及这些列的索引，Druid查询索引扫描段。段由数据源、间隔、版本的唯一标识，和一个可选的分区号。段命名规范如：datasource_interval_version_partitionnumber

    Segment sampleData_2011-01-01T01:00:00:00Z_2011-01-01T02:00:00:00Z_v1_0 contains

 2011-01-01T01:00:00Z  ultratrimfast.com  google.com  Male USA 1800  25  15.70
 2011-01-01T01:00:00Z  bieberfever.com    google.com  Male USA 2912  42  29.18

    Segment sampleData_2011-01-01T02:00:00:00Z_2011-01-01T03:00:00:00Z_v1_0 contains

 2011-01-01T02:00:00Z  ultratrimfast.com  google.com  Male UK  1953  17  17.31
 2011-01-01T02:00:00Z  bieberfever.com    google.com  Male UK  3194  170 34.01

    Druid数据预聚合：

• 在Druid查询的时候，Druid可以对摄入的数据事件按照指定的维度进行预聚合，预聚合可以极大地减少需要存储的数据的大小。

• Durid支持两种聚合方式，perfect（完美） roll-up 和 best-effort（最佳努力） roll-up。

• perface roll-up：完美的卷起模式包括一个额外的预处理步骤，这个预处理步骤通常扫描整个输入数据，这可能会增加摄取时间。Hadoop索引任务总是以这种完美的卷起模式运行。

• best-effort roll-up：最佳努力卷起模式不需要任何预处理步骤，但摄取数据的大小可能大于完美卷起的大小。所有类型的流索引（即RealTimeCindex任务，卡夫卡索引服务，…）都以这种模式运行。

rowId timestamp            publisher         advertiser  gender country impressions clicks revenue
1     2011-01-01T01:00:00Z ultratrimfast.com google.com  Male   USA     1800        25     15.70
2     2011-01-01T01:00:00Z bieberfever.com   google.com  Male   USA     2912        42     29.18
3     2011-01-01T02:00:00Z ultratrimfast.com google.com  Male   UK      1953        17     17.31
4     2011-01-01T02:00:00Z bieberfever.com   google.com  Male   UK      3194        170    34.01

    Druid数据索引：

• Druid查询速度取决于如何存储数据。从搜索基础架构借用想法，Druid创建只读数据快照，查询分析存储在高度优化的数据结构。

• Druid是一个列存储，每列被单独存储。Druid查询相当好，是因为只查询所需的列。不同的列还可以采用不同的压缩方式，不同的列也可以有与它们相关的不同的索引。

• Druid 索引数据在数据分片级别上。
• 对于聚合后的数据，如何对表建立索引，快速的进行查找我们知道数据库常用的索引是利用B+树建立联合索引，但是在以上数据中，比如按gender进行查找，由于该列的基数非常低，这样无论该表有多少行，B+数的叶子结点都比较少，所以查找索引的效率很低，并不一定比得上全表扫描。那么如何通过建立索引来解决这类问题呢？答案是建立位图索引。

• 位图索引

  key/rowId	1	2	3	4
  ultratrimfast.com	1	0	1	0
  bieberfever.com	0	1	0	1
  google.com	1	1	1	1
  Male	1	1	1	1
  USA	1	1	0	0
  UK	0	0	1	1

• 查询解码：比如查询gender=Male的广告点击次数，找到位图索引为1111，对应的行号为1，2，3，4行，然后将这几行的click列的值进行累加就得到了男性点击广告的次数。

    Druid数据摄入：

• Druid提供两种数据摄入的方式，实时数据摄入和批量数据摄入。

• Tranquility-Kafka：Tranquility的所有操作都是尽最大努力(best-effort)，我们可以通过配置多个任务副本保证数据不丢失，但是在某些情况下，数据可能会丢失或出现重复：（1）早于或晚于时间窗口，数据一定会被丢弃。（2）失败的Middle Manager数目多于配置的任务副本数，部分数据可能会丢失。（3）IndexingService内部（Overlord、MiddleManager、Peon）通信长时间丢失，同时重试次数超过最大上限，或者运行周期已经超过了时间窗口，这种情况部分数据也会被丢弃。（4）Tranquility一直未收到IndexingService的确认请求，那么Tranquility会切换到批量加载模式，数据可能会出现重复。所以，Druid官方建议，如果使用Tranquility作为Real-TimeNodes，那么可以采用如下解决方案减少数据丢失或者重复的风险，从而保证Druid中数据的exactly-oncesemantics：（1）将数据备份到S3或者HDFS等存储中；（2）晚间对备份数据运行Hadoopbatchindexingjob，从而对白天的数据重做Segment。
• Kafka-Indexing-Service：采用Kafkaindexingservice主要有以下几方面的考虑：（1）每一个进入Kafka的message都是有序、不变的，同时可以通过partition+offset的方式定位，而Druid作为Kafka的Consumer，可以通过该方式rewind到Kafka已存在的buffer中的任意一条message；（2）Message是由Consumer端，也就是Druid自主地pull进入，而不是被KafkaBrokerpush进集群，push的方式我们知道，接收端无法控制接收速率，容易造成数据过载，而pull的方式Consumer端可以控制ingest速率，从而保证数据有序、稳定地进入Druid；（3）Message中都包含了partition+offset标签，这就保证了作为Consumer的Druid可以通过确认机制保证每一条message都被读取，不会“被丢失”或“被重读”。
  所以，在Kafkaindexingservice中，每一个IndexingServiceTask都对应当前topic的一个partition，每一个partition都有对应的起止offset，那么Druid只需要按照offset顺序遍历读取该partition中所有的数据即可。同时，在读取过程中，Druid收到的每条message都会被确认，从而保证所有数据都被有序的读取，作索引，“加工成”Segment。当到达SegmentGranularity时，当前partition被读过的offset会被更新到元信息库的druid_dataSource表中。

2.Durid集群

    Druid集群节点：

• 1. Historica：对“historical”数据（非实时）进行处理存储和查询的地方。historical节点响应从broker节点发来的查询，并将结果返回给broker节点。它们在Zookeeper的管理下提供服务，并使用Zookeeper监视信号加载或删除新数据段。

• 2. Realtime：实时摄取数据，它们负责监听输入数据流并让其在内部的Druid系统立即获取，Realtime节点同样只响应broker节点的查询请求，返回查询结果到broker节点。旧数据会被从Realtime节点转存至Historical节点。

• 3. Coordinator：监控historical节点组，以确保数据可用、可复制，并且在一般的“最佳”配置。它们通过从MySQL读取数据段的元数据信息，来决定哪些数据段应该在集群中被加载，使用Zookeeper来确定哪个historical节点存在，并且创建Zookeeper条目告诉historical节点加载和删除新数据段。

• 4. Broker：接收来自外部客户端的查询，并将这些查询转发到Realtime和Historical节点。当Broker节点收到结果，它们将合并这些结果并将它们返回给调用者。由于了解拓扑，Broker节点使用Zookeeper来确定哪些Realtime和Historical节点的存在。

• 5. Indexer：节点会形成一个加载批处理和实时数据到系统中的集群，同时会对存储在系统中的数据变更（也称为索引服务）做出响应。这种分离让每个节点只关心自身的最优操作。通过将Historical和Realtime分离，将对进入系统的实时流数据监控和处理的内存分离。通过将Coordinator和Broker分离，把查询操作和维持集群上的“好的”数据分布的操作分离。

    Druid集群节点调用关系：

 

• 查询路径：红色箭头:①客户端向Broker发起请求,Broker会将请求路由到②实时节点和③历史节
• Druid数据流转:黑色箭头：数据源包括实时流和批量数据. ④实时流经过索引直接写到实时节点，⑤批量数据通过IndexService存储到DeepStorage,⑥再由历史节点加载. ⑦实时节点也可以将数据转存到DeepStorage。
• Druid的集群依赖了ZooKeeper来维护数据拓扑. 每个组件都会与ZooKeeper交互，如下：

 

• 实时节点在转存Segment到DeepStorage, 会写入自己转存了什么Segment
• 协调节点管理历史节点,它负责从ZooKeeper中获取要同步/下载的Segment,并指派任务给具体的历史节点去完成
• 历史节点从ZooKeeper中领取任务,任务完成后要将ZooKeeper条目删除表示完成了任务
• Broker节点根据ZooKeeper中的Segment所在的节点, 将查询请求路由到指定的节点
• 对于一个查询路由路径,Broker只会将请求分发到实时节点和历史节点, 因此元数据存储和DeepStorage都不会参与查询中(看做是后台的进程).

 Druid中MetaData Storage 与 Zookeeper：

• MetaStore和ZooKeeper中保存的信息是不一样的. ZooKeeper中保存的是Segment属于哪些节点. 而MetaStore则是保存Segment的元数据信息
• 为了使得一个Segment存在于集群中,MetaStore存储的记录是关于Segment的自描述元数据: Segment的元数据,大小,所在的DeepStorage
• 元数据存储的数据会被协调节点用来知道集群中可用的数据应该有哪些(Segment可以通过实时节点转存或者批量数据直接写入).

 

Druid还依赖于外部的三个组件：ZooKeeper, Metadata Storage, Deep Storage，数据与查询流的交互图如下：

 

• ① 实时数据写入到实时节点,会创建索引结构的Segment.
• ② 实时节点的Segment经过一段时间会转存到DeepStorage
• ③ 元数据写入MySQL; 实时节点转存的Segment会在ZooKeeper中新增一条记录
• ④ 协调节点从MySQL获取元数据,比如schema信息(维度列和指标列)
• ⑤ 协调节点监测ZK中有新分配/要删除的Segment,写入ZooKeeper信息:历史节点需要加载/删除Segment
• ⑥ 历史节点监测ZK, 从ZooKeeper中得到要执行任务的Segment
• ⑦ 历史节点从DeepStorage下载Segment并加载到内存/或者将已经保存的Segment删除掉
• ⑧ 历史节点的Segment可以用于Broker的查询路由
• 由于各个节点和其他节点都是最小化解耦的, 所以下面两张图分别表示实时节点和批量数据的流程:

•  数据从Kafka导入到实时节点, 客户端直接查询实时节点的数据。

•  批量数据使用IndexService,接收Post请求的任务,直接产生Segment写到DeepStorage里.DeepStorage中的数据只会被历史节点使用.所以这里要启动的服务有: IndexService(overlord), Historical, Coordinator(协调节点通知历史节点下载Segment)