druid 的基礎架構與應用

本文介紹了druid的基礎架構以及工作過程，通過一個應用案例加深了解。

durid簡介

druid是一種高性能、列式存儲、分散式數據存儲的時序數據分析引擎。能支持「PB」級數據的秒級查詢。類似的產品有kylin/clickhouse。druid典型的應用就是OLAP場景下的cube組合查詢分析。如數據鑽取（Drill-down）、上卷（Roll-up）、切片（Slice）、切塊（Dice）以及旋轉（Pivot）。後面的應用示例章節再詳細闡述。

durid基礎架構

先來了解一下durid主要節點：

1、broker node(代理節點)

Broker節點扮演著歷史節點和實時節點的查詢路由的角色。主要負責接收外部查詢，轉發查詢至各個segment數據所在的節點，並聚合結果返回。

2、historical node(歷史節點)

historical主要負責歷史數據存儲和查詢，接收協調節點數據載入與刪除指令，從deepstoage中下載segment，完成數據載入或者刪除後在zk中進行通告。歷史節點遵循shared-nothing的架構，因此節點間沒有單點問題。節點間是相互獨立的並且提供的服務也是簡單的，它們只需要知道如何載入、刪除和處理不可變的segment。historical節點也可以進行分組，組合成不同的historical tier。這會在集群規模較大的時候體現出優勢。如做數據的冷熱分離，按不同業務的數據分離（一定程度的資源隔離）。當然，historical 節點是整個集群查詢性能的核心所在，因為historical會承擔絕大部分的segment查詢。

3、coordinator node(協調節點)

主要負責數據的管理和在歷史節點上的分布。協調節點告訴歷史節點載入新數據、卸載過期數據、複製數據、和為了負載均衡移動數據。可以配置load數據及drop數據規則。

4、overlord node(index service 可以理解為任務管理節點）

功能描述：負責接收任務，管理任務。接收外部http請求（新建任務、查詢任務狀態、kill任務等），分配管理任務（當有新的任務請求，overload node會將任務分配給middleManager node去執行）。

5、middleManager node（可以理解為overlord節點的工作節點）

功能描述：可以啟動n（可配置）個peon，接收overlord分配的task，再交給自己peon去執行。

查詢過程

見上圖藍色箭頭，Broker節點接收到查詢（Q1），再將查詢發送給歷史節點與實時節點（Q2，Q3），在上圖的模式中，實時節點是MM節點上啟動的task。該task會負責數據的攝入以及提供實時數據的查詢。

數據攝入過程

見上圖紅色箭頭，D1是client生產數據最終寫入kafka（這個過程可能在client與kafka的中間，還包含了多個環節，如數據傳輸與數據清洗），D2和D3過程是部署tranquility-kafka服務，消費kafka數據寫入對應的task，tranquility-kakfa啟動的時候會跟overlord節點通信，由overlord節點分配任務給middleManager執行。D4是task 負責的segment段正常結束，然後將segment數據寫入deepstorage過程。（實時task運行時間是segmentGranularity+windowPeriod+intermediatePersistPeriod）。D5則是historical節點從deepstorage下載segment並在zk中聲明負責該segment段查詢的過程。

目前druid數據攝入過程還有一種更推薦的方式就是kafka index service(簡稱kis)，有興趣的同學可以參考官方文檔，kis對kafka的版本有強要求。

druid整體架構雖然略為複雜，但是整體穩定性非常不錯，幾乎很少出現集群故障。拋開集群硬體故障和數據本身問題，SLA基本能到4個9。coordinator，overlord兩個節點是主從模式，保證每個角色起兩個實例即可。broker節點無狀態，可以起多個實例，前面掛個域名即可（為了保證緩存命中，最好配置ip hash）。historical節點無狀態，有一定冗餘即可。middleManager用作數據攝入節點，若task沒有配置副本，則節點宕機會引發丟數據的風險。當然，kis可以避免該問題。

durid數據聚合、存儲核心思想

druid 數據存儲分為三部分timestamp、dimensions、metrics。其中，timestamp、metrics部分是採用lz4直接壓縮。

但是dimensions部分需要支持過濾查詢以及分組查詢。所以dimensions部分的每個維度都採用了以下三種數據結構做轉碼、存儲：

1. A dictionary that maps values (which are always treated as strings) to integer IDs，
2. For each distinct value in the column，a bitmap that indicates which rows contain that value，and
3. A list of the column』s values，encoded using the dictionary in 1

舉個例子，源數據如下：

name列來說

1. Dictionary that encodes column values

字典表的key都是唯一的，所以Map的key是unique的column value，Map的value從0開始不斷增加。 示例數據的name列只有兩個不同的值。所以張三編號為0，李四編號為1：

{
"張三": 0
"李四": 1
}

2. Column data

要保存的是每一行中這一列的值，值是ID而不是原始的值。因為有了上面的Map字典，所以有下面的對應關係:

[0，

1，

1，

0]

3. Bitmaps - one for each unique value of the column

BitMap的key是第一步Map的key(原始值)， value則是真假的一個標識（是|否？等於|不等於？），取值只有0、1，如下：

value="張三": [1,0,0,1]

value=「李四": [0,1,1,0]

所以由上可知最壞的情況可能是隨著數據量的增加，bitmap的個數也成線性增長，為數據量大小*列的個數。那麼在什麼情況下會導致這種線性增長？這裡我們引入了一個基數(cardinality)的概念。基數=unique（dim1,dim2.....），如若dim取值均為各種爆炸性id或者隨機數，則druid的預聚合將完全失去意義。所以在druid的應用場景中，基數約小，聚合效率越高。

講了dimensions怎麼存儲，那麼metrics又是怎麼聚合（roll-up）呢？這就要引入druid數據schema定義了。下一章結合應用一塊看一個示例。

應用示例與實踐經驗

假設有這樣一份數據，典型的商品銷售數據。

我們構造成druid中的數據schema如下：

{
"dataSources" : [ {
"spec" : {
"dataSchema" : {
"dataSource" : "test_datasource",
"granularitySpec" : {
"segmentGranularity" : "hour",
"queryGranularity" : "minute",
"type" : "uniform"
},
"parser" : {
"type" : "string",
"parseSpec" : {
"format" : "json",
"timestampSpec" : {
"column" : "time",
"format" : "auto"
},
"dimensionsSpec" : {
"dimensions" : [ "productName", "city", "channel", 「action"]
}
}
},
"metricsSpec" : [ {
"name" : "count",
"type" : "count"
}, {
"type" : "doubleSum",
"fieldName" : "price",
"name" : 「sale"
} ]
},
"tuningConfig" : {
"type" : "realtime",
"windowPeriod" : "PT10M",
"intermediatePersistPeriod" : "PT10M",
"maxRowsInMemory" : "100000"
}
},
"properties" : {
"topicPattern" : "test_datasource",
"task.partitions" : "2",
"task.replicants" : "1"
}
} ],
"properties" : {
...
}
}

前面重點說了dimensions，我們再來看下metrics。在上面的例子中我們只定義count和針對price的doubleSum，那麼這些指標就已經固定了後期的分析需求。我們看到上面table中的一二行標紅部分，所有dim取值完全相同，queryGranularity為一分鐘。那麼在這2018-06-11 12:23:00這個點，這兩行數據就被聚合成一行，count=2，sale=0。以此類推。

然後我們再來看看具體的分析需求，一個鑽取的例子。我們首先查看商品A昨天的點擊量，select sum(count) from table where productName=『A』 and action=『click"，再想看看地區=北京，渠道=web呢？是不是再加幾個where就搞定了？select sum(count) from table where productName=『A』 and city=『北京』 and channel=『web" and action=『click』; 然後就是切片和切塊，也很簡單，就是幾個group by。這些在druid中都能非常輕鬆的支持。

具體使用上的經驗總結：

1. reindex思想。一般我們實時數據查詢粒度配置的會比較小，秒級或者分鐘級。那麼對於一天前，三天前，一個月前的數據呢？這時候一般關注的粒度將不再那麼細，所以我們一般會採取redinx的策略進行再聚合

2. 針對歷史數據，可能對於某些維度將不在關心，這時候我們也可以在reindex時，將無用的維度剔除掉，可能大大減少整體數據的基數。

3. 一般數據壓縮比例。這裡提供一個大概的參考值。數據總基數在10W以下，每天數據量約百億左右，druid中聚合後的索引數據與原始數據大小之比可以到1：100，甚至1：1000。

4. druid適用於常規的olap場景，能非常輕鬆的支撐每天百億甚至千億級別的數據寫入。

5. 爆炸性維度數據，以及頻繁update數據的需求，不適用於druid的場景。

總結

本文主要對druid做了入門級的基礎介紹，可以給大家做olap引擎技術選型時做一個參考。以及對druid的初學者做一個大致介紹。druid是一款非常優秀的olap引擎，從性能、穩定性上來說，都是非常不錯的。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！