1.什么是Druid？

Druid是一個(gè)高效的數(shù)據(jù)查詢系統(tǒng)，主要解決的是對于大量的基于時(shí)序的數(shù)據(jù)進(jìn)行聚合查詢。數(shù)據(jù)可以實(shí)時(shí)攝入，進(jìn)入到Druid后立即可查，同時(shí)數(shù)據(jù)是幾乎是不可變。通常是基于時(shí)序的事實(shí)事件，事實(shí)發(fā)生后進(jìn)入Druid，外部系統(tǒng)就可以對該事實(shí)進(jìn)行查詢。
Druid采用的架構(gòu):
shared-nothing架構(gòu)與lambda架構(gòu)
Druid設(shè)計(jì)三個(gè)原則:
1.快速查詢（Fast Query） : 部分?jǐn)?shù)據(jù)聚合（Partial Aggregate） + 內(nèi)存華（In-Memory） + 索引（Index）
2.水平拓展能力（Horizontal Scalability）:分布式數(shù)據(jù)（Distributed data）+并行化查詢（Parallelizable Query）
3.實(shí)時(shí)分析（Realtime Analytics）：Immutable Past , Append-Only Future

2.Druid的技術(shù)特點(diǎn)

數(shù)據(jù)吞吐量大
支持流式數(shù)據(jù)攝入和實(shí)時(shí)
查詢靈活且快速

3.Druid基本概念：

Druid在數(shù)據(jù)攝入之前，首先要定義一個(gè)數(shù)據(jù)源（DataSource,類似于數(shù)據(jù)庫中表的概念）
Druid是一個(gè)分布式數(shù)據(jù)分析平臺,也是一個(gè)時(shí)序數(shù)據(jù)庫
1.數(shù)據(jù)格式
數(shù)據(jù)集合（時(shí)間列，維度列，指標(biāo)列）
數(shù)據(jù)結(jié)構(gòu)：
基于DataSource與Segment的數(shù)據(jù)結(jié)構(gòu)，DataSource相當(dāng)于關(guān)系型數(shù)據(jù)庫中的表。
DataSource包含：
時(shí)間列（TimeStamp）：標(biāo)識每行數(shù)據(jù)的時(shí)間值
維度列（Dimension）：標(biāo)識數(shù)據(jù)行的各個(gè)類別信息
指標(biāo)列（Metric）:用于聚合和計(jì)算的列
Segment結(jié)構(gòu)：
DataSource是邏輯結(jié)構(gòu)，而Segment是數(shù)據(jù)實(shí)際存儲的物理結(jié)構(gòu)，Druid通過Segment實(shí)現(xiàn)對數(shù)據(jù)的橫縱切割操作。
橫向切割：通過設(shè)置在segmentGranularity參數(shù)，Druid將不同時(shí)間范圍內(nèi)的數(shù)據(jù)存儲在不同Segment數(shù)據(jù)塊中。
縱向切割：在Segment中面向列進(jìn)行數(shù)據(jù)壓縮處理

設(shè)置合理的Granularity
segmentGranularity：segment的組成粒度。
queryGranularity ：segment的聚合粒度。
queryGranularity 小于等于 segmentGranularity
若segmentGranularity = day，那么Druid會按照天把不同天的數(shù)據(jù)存儲在不同的Segment中。
若queryGranularity =none,可以查詢所有粒度，queryGranularity = hour只能查詢>=hour粒度的數(shù)據(jù)

2.數(shù)據(jù)攝入
實(shí)時(shí)數(shù)據(jù)攝入
批處理數(shù)據(jù)攝入

3.數(shù)據(jù)查詢
原生查詢，采用JSON格式，通過http傳送

4.時(shí)序數(shù)據(jù)庫

1.OpenTSDB
開源的時(shí)序數(shù)據(jù)庫，支持?jǐn)?shù)千億的數(shù)據(jù)點(diǎn)，并提供精確的數(shù)據(jù)查詢功能
采用java編寫，通過基于Hbase的存儲實(shí)現(xiàn)橫向拓展
設(shè)計(jì)思路：利用Hbase的key存儲一些tag信息，將同一小時(shí)的數(shù)據(jù)放在一行存儲，提高了查詢速度
架構(gòu)示意圖：

2.Pinot
接近Druid的系統(tǒng)
Pinot也采用了Lambda架構(gòu)，將實(shí)時(shí)流和批處理數(shù)據(jù)分開處理
Realtime Node處理實(shí)時(shí)數(shù)據(jù)查詢
Historical Node處理歷史數(shù)據(jù)
技術(shù)特點(diǎn)：
面向列式存儲的數(shù)據(jù)庫，支持多種壓縮技術(shù)
可插入的索引技術(shù) — Sorted index ，Bitmap index， Inverted index
可以根據(jù)Query和Segment元數(shù)據(jù)進(jìn)行查詢和執(zhí)行計(jì)劃的優(yōu)化
從kafka實(shí)時(shí)灌入數(shù)據(jù)和從hadoop的批量數(shù)據(jù)灌入
類似于SQL的查詢語言和各種常用聚合
支持多值字段
水平拓展和容錯(cuò)
Pinot架構(gòu)圖：

3.Druid架構(gòu)概覽

Druid包含以下四個(gè)節(jié)點(diǎn)：
實(shí)時(shí)節(jié)點(diǎn)（Realtime ）:即時(shí)攝入實(shí)時(shí)數(shù)據(jù)，以及生成Segment數(shù)據(jù)文件
實(shí)時(shí)節(jié)點(diǎn)負(fù)責(zé)消費(fèi)實(shí)時(shí)數(shù)據(jù)，實(shí)時(shí)數(shù)據(jù)首先會被直接加載進(jìn)實(shí)時(shí)節(jié)點(diǎn)內(nèi)存中的堆結(jié)構(gòu)緩存區(qū)，當(dāng)條件滿足時(shí)，
緩存區(qū)的數(shù)據(jù)會被沖寫到硬盤上形成一個(gè)數(shù)據(jù)塊（Segment Split），同時(shí)實(shí)時(shí)節(jié)點(diǎn)又會立即將新生成的數(shù)據(jù)庫加載到內(nèi)存的非堆區(qū)，
因此無論是堆結(jié)構(gòu)緩存區(qū)還是非堆區(qū)里的數(shù)據(jù)都能被查詢節(jié)點(diǎn)（Broker Node）查詢
歷史節(jié)點(diǎn)(Historical Node):加載已經(jīng)生成好的文件，以供數(shù)據(jù)查詢
查詢節(jié)點(diǎn)(Broker Node)：對外提供數(shù)據(jù)查詢服務(wù)，并同時(shí)從實(shí)時(shí)節(jié)點(diǎn)和歷史節(jié)點(diǎn)查詢數(shù)據(jù)，合并后返回給調(diào)用方
協(xié)調(diào)節(jié)點(diǎn)（Coordinator Node）：負(fù)責(zé)歷史節(jié)點(diǎn)的數(shù)據(jù)負(fù)載均衡，以及通過規(guī)則（Rule）管理數(shù)據(jù)的生命周期

集群還依賴三類外部依賴
元數(shù)據(jù)庫（Metastore）：存儲Druid集群的原數(shù)據(jù)信息，比如Segment的相關(guān)信息，一般用MySql或PostgreSQL
分布式協(xié)調(diào)服務(wù)（Coordination）:為Druid集群提供一致性協(xié)調(diào)服務(wù)的組件，通常為Zookeeper
數(shù)據(jù)文件存儲系統(tǒng)（DeepStorage）：存放生成的Segment文件，并供歷史節(jié)點(diǎn)下載。對于單節(jié)點(diǎn)集群可以是本地磁盤，而對于分布式集群一般是HDFS或NFS

實(shí)時(shí)節(jié)點(diǎn)數(shù)據(jù)塊的生成示意圖：

數(shù)據(jù)塊的流向：

Realtime Node 實(shí)時(shí)節(jié)點(diǎn)：
1.通過Firehose來消費(fèi)實(shí)時(shí)數(shù)據(jù)，F(xiàn)irehose是Druid中消費(fèi)實(shí)時(shí)數(shù)據(jù)的模型
2.實(shí)時(shí)節(jié)點(diǎn)會通過一個(gè)用于生成Segment數(shù)據(jù)文件的模塊Plumber（具體實(shí)現(xiàn)有RealtimePlumber等）按照指定的周期，按時(shí)將本周起生產(chǎn)的所有數(shù)據(jù)塊合并成一個(gè)大的Segment文件

Historical Node歷史節(jié)點(diǎn)

歷史節(jié)點(diǎn)在啟動(dòng)的時(shí)候 ：
1、會去檢查自己本地緩存中已經(jīng)存在的Segment數(shù)據(jù)文件
2、從DeepStorege中下載屬于自己但是目前不在自己本地磁盤上的Segment數(shù)據(jù)文件
無論何種查詢，歷史節(jié)點(diǎn)會首先將相關(guān)Segment數(shù)據(jù)文件從磁盤加載到內(nèi)存，然后在提供查詢

Broker Node節(jié)點(diǎn)：
Druid提供兩類介質(zhì)作為Cache以供選擇
外部Cache,比如Memcached
本地Cache,比如查詢節(jié)點(diǎn)或歷史節(jié)點(diǎn)的內(nèi)存作為cache

高可用性：
通過Nginx來負(fù)載均衡查詢節(jié)點(diǎn)（Broker Node）

協(xié)調(diào)節(jié)點(diǎn)：
協(xié)調(diào)節(jié)點(diǎn)（Coordinator Node）負(fù)責(zé)歷史節(jié)點(diǎn)的數(shù)據(jù)負(fù)載均衡，以及通過規(guī)則管理數(shù)據(jù)的生命周期

4.索引服務(wù)

1.其中主節(jié)點(diǎn)：overlord
兩種運(yùn)行模式：
本地模式（Local Mode）：默認(rèn)模式，主節(jié)點(diǎn)負(fù)責(zé)集群的任務(wù)協(xié)調(diào)分配工作，也能夠負(fù)責(zé)啟動(dòng)一些苦工（Peon）來完成一部分具體任務(wù)
遠(yuǎn)程模式（Remote）：該模式下，主節(jié)點(diǎn)與從節(jié)點(diǎn)運(yùn)行在不同的節(jié)點(diǎn)上，它僅負(fù)責(zé)集群的任務(wù)協(xié)調(diào)分配工作，不負(fù)責(zé)完成具體的任務(wù)，主節(jié)點(diǎn)提供RESTful的訪問方法，因此客戶端可以通過HTTP POST
請求向主節(jié)點(diǎn)提交任務(wù)。
命令格式如下：
http://<ioverlord_ip>:port/druid/indexer/v1/task
刪除任務(wù)：http://<ioverlord_ip>:port/druid/indexer/v1/task/{taskId}/shutdown
控制臺：http://<ioverlord_ip>:port/console.html

2.從節(jié)點(diǎn)：Middle Manager
索引服務(wù)的工作節(jié)點(diǎn)，負(fù)責(zé)接收主節(jié)點(diǎn)的分配的任務(wù)，然后啟動(dòng)相關(guān)的苦工即獨(dú)立的JVM來完成具體的任務(wù)
這樣的架構(gòu)與Hadoop YARN相似
主節(jié)點(diǎn)相當(dāng)于Yarn的ResourceManager,負(fù)責(zé)集群資源管理，與任務(wù)分配
從節(jié)點(diǎn)相當(dāng)于Yarn的NodeManager,負(fù)責(zé)管理獨(dú)立節(jié)點(diǎn)的資源并接受任務(wù)
Peon(苦工)相當(dāng)于Yarn的Container,啟動(dòng)在具體節(jié)點(diǎn)上負(fù)責(zé)具體任務(wù)的執(zhí)行

問題：

由于老版本的Druid使用pull方式消費(fèi)kafka數(shù)據(jù)，使用kafka consumer group來共同消費(fèi)一個(gè)kafka topic的數(shù)據(jù)，各個(gè)節(jié)點(diǎn)會負(fù)責(zé)獨(dú)立消費(fèi)一個(gè)或多個(gè)該topic所包含的Partition數(shù)據(jù)，并保證同一個(gè)Partition不會被多于一個(gè)的實(shí)時(shí)節(jié)點(diǎn)消費(fèi)。每當(dāng)一個(gè)實(shí)時(shí)節(jié)點(diǎn)完成部分?jǐn)?shù)據(jù)的消費(fèi)后，會主動(dòng)將消費(fèi)進(jìn)度（kafka topic offset）提交到Zookeeper集群。
當(dāng)節(jié)點(diǎn)不可用時(shí)，該kafka consumer group 會立即在組內(nèi)對所有可用的節(jié)點(diǎn)進(jìn)行partition重新分配，接著所有節(jié)點(diǎn)將會根據(jù)記錄在zk集群中每一個(gè)partition的offset來繼續(xù)消費(fèi)未曾消費(fèi)的數(shù)據(jù)，從而保證所有數(shù)據(jù)在任何時(shí)候都會被Druid集群至少消費(fèi)一次。
這樣雖然能保證不可用節(jié)點(diǎn)未消費(fèi)的partition會被其余工作的節(jié)點(diǎn)消費(fèi)掉，但是不可用節(jié)點(diǎn)上已經(jīng)消費(fèi)的數(shù)據(jù)，尚未被傳送到DeepStoreage上且未被歷史節(jié)點(diǎn)下載的Segment數(shù)據(jù)卻會被集群遺漏，這是基于kafka-eight Firehose消費(fèi)方式的一種缺陷。
解決方案：
1.讓不可用節(jié)點(diǎn)恢復(fù)重新回到集群成為可用節(jié)點(diǎn)，重啟后會將之前已經(jīng)生成但未上傳的Segment數(shù)據(jù)文件統(tǒng)統(tǒng)加載回來，并最終合并傳送到DeepStoreage，保證數(shù)據(jù)完整性
2.使用Tranquility與Indexing Service,對kafka的數(shù)據(jù)進(jìn)行精確的消費(fèi)與備份。
由于Tranquility可以通過push的方式將指定數(shù)據(jù)推向Druid集群，因此它可以同時(shí)對同一個(gè)partition制造多個(gè)副本。所以當(dāng)某個(gè)數(shù)據(jù)消費(fèi)失敗時(shí)候，系統(tǒng)依然可以準(zhǔn)確的選擇使用另外一個(gè)相同的任務(wù)所創(chuàng)建的Segment數(shù)據(jù)庫