Druid.io系列（三）： Druid集羣節點

原文連接: https://blog.csdn.net/njpjsoftdev/article/details/52955937

 

1 Historical Node

Historical Node的職責單一，就是負責加載Druid中非實時窗口內且知足加載規則的全部歷史數據的Segment。每個Historical Node只與Zookeeper保持同步，不與其餘類型節點或者其餘Historical Node進行通訊。

根據上節知曉，Coordinator Nodes會按期（默認爲1分鐘）去同步元信息庫，感知新生成的Segment，將待加載的Segment信息保存在Zookeeper中在線的Historical Nodes的load queue目錄下，當Historical Node感知到須要加載新的Segment時，首先會去本地磁盤目錄下查找該Segment是否已下載，若是沒有，則會從Zookeeper中下載待加載Segment的元信息，此元信息包括Segment存儲在何處、如何解壓以及如何如理該Segment。Historical Node使用內存文件映射方式將index.zip中的XXXXX.smoosh文件加載到內存中，並在Zookeeper中本節點的served segments目錄下聲明該Segment已被加載，從而該Segment能夠被查詢。對於從新上線的Historical Node，在完成啓動後，也會掃描本地存儲路徑，將全部掃描到的Segment加載如內存，使其可以被查詢。

2 Broker Node

Broker Node是整個集羣查詢的入口，做爲查詢路由角色，Broker Node感知Zookeeper上保存的集羣內全部已發佈的Segment的元信息，即每一個Segment保存在哪些存儲節點上，Broker Node爲Zookeeper中每一個dataSource建立一個timeline，timeline按照時間順序描述了每一個Segment的存放位置。咱們知道，每一個查詢請求都會包含dataSource以及interval信息，Broker Node根據這兩項信息去查找timeline中全部知足條件的Segment所對應的存儲節點，並將查詢請求發往對應的節點。

對於每一個節點返回的數據，Broker Node默認使用LRU緩存策略；對於集羣中存在多個Broker Node的狀況，Druid使用memcached共享緩存。對於Historical Node返回的結果，Broker Node認爲是「可信的」，會緩存下來，而Real-Time Node返回的實時窗口內的數據，Broker Node認爲是可變的，「不可信的」，故不會緩存。因此對每一個查詢請求，Broker Node都會先查詢本地緩存，若是不存在纔會去查找timeline，再向相應節點發送查詢請求。

3 Coordinator Node

Coordinator Node主要負責Druid集羣中Segment的管理與發佈，包括加載新Segment、丟棄不符合規則的Segment、管理Segment副本以及Segment負載均衡等。若是集羣中存在多個Coordinator Node，則經過選舉算法產生Leader，其餘Follower做爲備份。

Coordinator會按期（默認一分鐘）同步Zookeeper中整個集羣的數據拓撲圖、元信息庫中全部有效的Segment信息以及規則庫，從而決定下一步應該作什麼。對於有效且未分配的Segment，Coordinator Node首先按照Historical Node的容量進行倒序排序，即最少容量擁有最高優先級，新的Segment會優先分配到高優先級的Historical Node上。由3.3.4.1節可知，Coordinator Node不會直接與Historical Node打交道，而是在Zookeeper中Historical Node對應的load queue目錄下建立待加載Segment的臨時信息，等待Historical Node去加載該Segment。

Coordinator在每次啓動後都會對比Zookeeper中保存的當前數據拓撲圖以及元信息庫中保存的數據信息，全部在集羣中已被加載的、卻在元信息庫中標記爲失效或者不存在的Segment會被Coordinator Node記錄在remove list中，其中也包括咱們在3.3.3節中所述的同一Segment對應的新舊version，舊version的Segments一樣也會被放入到remove list中，最終被邏輯丟棄。

對於離線的Historical Node，Coordinator Node會默認該Historical Node上全部的Segment已失效，從而通知集羣內的其餘Historical Node去加載該Segment。可是，在生產環境中，咱們會遇到機器臨時下線，Historical Node在很短期內恢復服務的狀況，那麼如此「簡單粗暴」的策略勢必會加劇整個集羣內的網絡負載。對於這種場景，Coordinator會爲集羣內全部已丟棄的Segment保存一個生存時間(lifetime)，這個生存時間表示Coordinator Node在該Segment被標記爲丟棄後，容許不被從新分配最長等待時間，若是該Historical Node在該時間內從新上線，則Segment會被從新置爲有效，若是超過該時間則會按照加載規則從新分配到其餘Historical Node上。

考慮一種最極端的狀況，若是集羣內全部的Coordinator Node都中止服務，整個集羣對外依然有效，不過新Segment不會被加載，過時的Segment也不會被丟棄，即整個集羣內的數據拓撲會一直保持不變，直到新的Coordinator Node服務上線。

4 Indexing Service

Indexing Service是負責「生產」Segment的高可用、分佈式、Master/Slave架構服務。主要由三類組件構成：負責運行索引任務(indexing task)的Peon，負責控制Peon的MiddleManager，負責任務分發給MiddleManager的Overlord；三者的關係能夠解釋爲：Overlord是MiddleManager的Master，而MiddleManager又是Peon的Master。其中，Overlord和MiddleManager能夠分佈式部署，可是Peon和MiddleManager默認在同一臺機器上。圖3.5給出了Indexing Service的總體架構。

Overlord 
Overlord負責接受任務、協調任務的分配、建立任務鎖以及收集、返回任務運行狀態給調用者。當集羣中有多個Overlord時，則經過選舉算法產生Leader，其餘Follower做爲備份。

Overlord能夠運行在local（默認）和remote兩種模式下，若是運行在local模式下，則Overlord也負責Peon的建立與運行工做，當運行在remote模式下時，Overlord和MiddleManager各司其職，根據圖3.6所示，Overlord接受實時/批量數據流產生的索引任務，將任務信息註冊到Zookeeper的/task目錄下全部在線的MiddleManager對應的目錄中，由MiddleManager去感知產生的新任務，同時每一個索引任務的狀態又會由Peon按期同步到Zookeeper中/Status目錄，供Overlord感知當前全部索引任務的運行情況。

Overlord對外提供可視化界面，經過訪問http://:/console.html，咱們能夠觀察到集羣內目前正在運行的全部索引任務、可用的Peon以及近期Peon完成的全部成功或者失敗的索引任務。

MiddleManager 
MiddleManager負責接收Overlord分配的索引任務，同時建立新的進程用於啓動Peon來執行索引任務，每個MiddleManager能夠運行多個Peon實例。

在運行MiddleManager實例的機器上，咱們能夠在${ java.io.tmpdir}目錄下觀察到以XXX_index_XXX開頭的目錄，每個目錄都對應一個Peon實例；同時restore.json文件中保存着當前全部運行着的索引任務信息，一方面用於記錄任務狀態，另外一方面若是MiddleManager崩潰，能夠利用該文件重啓索引任務。

Peon 
Peon是Indexing Service的最小工做單元，也是索引任務的具體執行者，全部當前正在運行的Peon任務均可以經過Overlord提供的web可視化界面進行訪問。

5 Real-Time Node

在流式處理領域，有兩種數據處理模式，一種爲Stream Push，另外一種爲Stream Pull。

Stream Pull 
若是Druid以Stream Pull方式自主地從外部數據源拉取數據從而生成Indexing Service Tasks，咱們則須要創建Real-Time Node。Real-Time Node主要包含兩大「工廠」：一個是鏈接流式數據源、負責數據接入的Firehose（中文翻譯爲水管，很形象地描述了該組件的職責）；另外一個是負責Segment發佈與轉移的Plumber（中文翻譯爲搬運工，一樣也十分形象地描述了該組件的職責）。在Druid源代碼中，這兩個組件都是抽象工廠方法，使用者能夠根據本身的需求建立不一樣類型的Firehose或者Plumber。Firehose和Plumber給個人感受，更相似於Kafka_0.9.0版本後發佈的Kafka Connect框架，Firehose相似於Kafka Connect Source，定義了數據的入口，但並不關心接入數據源的類型；而Plumber相似於Kafka Connect Sink，定義了數據的出口，也不關心最終輸出到哪裏。

Stream Push 
若是採用Stream Push策略，咱們須要創建一個「copy service」，負責從數據源中拉取數據並生成Indexing Service Tasks，從而將數據「推入」到Druid中，咱們在druid_0.9.1版本以前一直使用的是這種模式，不過這種模式須要外部服務Tranquility，Tranquility組件能夠鏈接多種流式數據源，好比Spark-Streaming、Storm以及Kafka等，因此也產生了Tranquility-Storm、Tranquility-Kafka等外部組件。Tranquility-Kafka的原理與使用將在3.4節中進行詳細介紹。

6 外部拓展

Druid集羣依賴一些外部組件，與其說依賴，不如說正是因爲Druid開放的架構，因此用戶能夠根據本身的需求，使用不一樣的外部組件。

Deep Storage 
Druid目前支持使用本地磁盤(單機模式)、NFS掛載磁盤、HDFS、Amazon S3等存儲方式保存Segments以及索引任務日誌。

Zookeeper 
Druid使用Zookeeper做爲分佈式集羣內部的通訊組件，各種節點經過Curator Framework將實例與服務註冊到Zookeeper上，同時將集羣內須要共享的信息也存儲在Zookeeper目錄下，從而簡化集羣內部自動鏈接管理、leader選舉、分佈式鎖、path緩存以及分佈式隊列等複雜邏輯。

Metadata Storage 
Druid集羣元信息使用MySQL 或者PostgreSQL存儲，單機版使用derby。在Druid_0.9.1.1版本中，元信息庫druid主要包含十張表，均以「druid_」開頭，如圖3.7所示。 
 
 
 
 
 
 
 
 

7 加載數據

對於加載外部數據，Druid支持兩種模式：實時流(real-time ingestion)和批量導入(batch ingestion)。

Real-Time Ingestion 
實時流過程能夠採用Apache Storm、Apache Spark Streaming等流式處理框架產生數據，再通過pipeline工具，好比Apache Kafka、ActiveMQ、RabbitMQ等消息總線類組件，使用Stream Pull 或者Stream Push模式生成Indexing Service Tasks，最終存儲在Druid中。

Batch Ingestion 
批量導入模式能夠採用結構化信息做爲數據源，好比JSON、Avro、Parquet格式的文本，Druid內部使用Map/Reduce批處理框架導入數據。

8 高可用性

Druid高可用性能夠總結如下幾點：

Historical Node 
如3.3.4.1節中所述，若是某個Historical Node離線時長超過必定閾值，Coordinator Node會將該節點上已加載的Segments從新分配到其餘在線的Historical Nodes上，保證知足加載規則的全部Segments不丟失且可查詢。

Coordinator Node 
集羣可配置多個Coordinator Node實例，工做模式爲主從同步，採用選舉算法產生Leader，其餘Follower做爲備份。當Leader宕機時，其餘Follower可以迅速failover。 
即便當全部Coordinator Node均失效，整個集羣對外依然有效，不過新Segments不會被加載，過時的Segments也不會被丟棄，即整個集羣內的數據拓撲會一直保持不變，直到新的Coordinator Node服務上線。

Broker Node 
Broker Node與Coordinator Node在HA部署方面一致。

Indexing Service 
Druid能夠爲同一個Segment配置多個Indexing Service Tasks副本保證數據完整性。

Real-Time 
Real-Time過程的數據完整性主要由接入的實時流語義(semantics)決定。咱們在0.9.1.1版本前使用Tranquility-Kafka組件接入實時數據，因爲存在時間窗口，即在時間窗口內的數據會被提交給Firehose，時間窗口外的數據則會被丟棄；若是Tranquility-Kafka臨時下線，會致使Kafka中數據「過時」從而被丟棄，沒法保證數據完整性，同時這種「copy service」的使用模式不只佔用大量CPU與內存，又不知足原子操做，因此在0.9.1.1版本後，咱們使用Druid的新特性Kafka Indexing Service，Druid內部使用Kafka高級Consumer API保證exactly-once semantics，盡最大可能保證數據完整性。不過咱們在使用中，依然發現有數據丟失問題。

Metadata Storage 
若是Metadata Storage失效，Coordinator則沒法感知新Segment的生成，整個集羣中數據拓撲亦不會改變，不過不會影響老數據的訪問。

Zookeeper 
若是Zookeeper失效，整個集羣的數據拓撲不會改變，因爲Broker Node緩存的存在，因此在緩存中的數據依然能夠被查詢。

9 數據分層

Druid訪問控制策略採用數據分層(tier)，有如下兩種用途：

1. 將不一樣的Historical Node劃分爲不一樣的group，從而控制集羣內不一樣權限(priority)用戶在查詢時訪問不一樣group。

2. 經過劃分tier，讓Historical Node加載不一樣時間範圍的數據。例如tier_1加載2016年Q1數據，tier_2加載2016年Q2數據，tier_3加載2016年Q3數據等；那麼根據用戶不一樣的查詢需求，將請求發往對應tier的Historical Node，不只能夠控制用戶訪問請求，同時也能夠減小響應請求的Historical Node數量，從而加速查詢。