apache druid 實時加載kafka 中的數據（一）

簡介

apache druid 是分佈式列存儲的 OLAP 框架。仍是一個時間序列數據庫。本篇文章主要是druid 在kafka 加載數據的配置。因爲druid 升級狀況太快，本人的環境仍是在0.13，主要改動方面仍是UI，新的版本在UI方面更適合新手入門。

文章若有幫助，請關注微信公共號。

最終使用druid時，是0.9版本，當時在kafka加載數據推薦的方式是兩種

• Tranquility

• kafka index service

Tranquility

是用於將流實時推送到Druid的工具包。是一個獨立，須要單獨下載。

** 其特色**

無縫地處理分區，複製，服務發現和架構過渡，而無需停機。集成了http server,Samza，Spark ，Storm，Flink 等工具。

能夠自由的控制向druid，主動發送數據。

** 劣勢**

自己具備時間窗，超過期間窗的數據直接丟棄。

版本落後，因爲沒有官方組織維護，目前版本只是兼容值0.9.2，後面druid升級後，Tranquility未及時升級，有些新的api 沒法適配。

kafka index service

這是druid 自身攜帶的擴展插件，使用時，須要在common.runtime.properties 文件中的屬性 druid.extensions.loadList 添加druid-kafka-indexing-service。

**  其特色**

支持實時查詢按時間分segment,非實追加到對應時間的segment 。

經過算法把Peon分配到 不一樣的【 Middle Managers】上實現分佈式

加大對應kafka的topic的partition數量 加大taskCount的值，產生更多的Peon

建立 supervisor

上面是一個完整的supervisor的內容，主要包含type，dataSchema，tuningConfig，ioConfig 四個部分

• type

標記類型，supervisor  的類型 就是kafka.

• dataSchema

數據庫的配置，主要包含dataSource，parser，metricsSpec，granularitySpec

dataSource

druid的數據庫名稱。

parser

配置與解析數據。簡單理解就是kafka中的數據與druid存儲之間的關係映射。主要包含如下配置

timestampSpec

配置處於的位置 dataSchema->parser->timestampSpec

druid 自己是時間序列數據庫，故此時間就是數據的主鍵。因爲druid 在 0.9後，已經不支持設置時區了，時間都是採用的utc格式。druid查詢時，能夠設置時區。包括一些roll-up操做都是按照utc時間進行。若有必須需改動源碼。

dimensionsSpec

位置：dataSchema->parser->dimensionsSpec

維度。數據庫須要存儲的字段，須要與kafka中的對應。

dimensions

是一個數組類型，默認字段的類型都是string

能夠設置字段的類型，例如{ "type": "long", "name": "userId" }

metricsSpec

位置：dataSchema->metricsSpec

度量。此值roll-up 啓用是纔有意義。

`{      "name": "theta_customer_id",

"type": "thetaSketch",

"fieldName": "customer_id"
           } `

name: druid中字段的名稱。

type:指標類型。thetaSketch 去重。還支持doubleSum，longSum，doubleMin,doubleMax 等聚合類型。

fieldName：kafka中 屬性的名稱

granularitySpec

位置：      dataSchema->granularitySpec

segmentGranularity: Segment粒度（SegmentGranularity）表示每個實時索引任務中產生的Segment所涵蓋的時間範圍。

queryGranularity：查詢粒度。例如 {"queryGranularity":"DAY"} 查詢的最小粒度就是DAY,通過roll-up後，維度徹底同樣的數據，一天範圍內將聚合爲一條數據。

• tuningConfig

調優相關的配置。

配置一個segment大小。

調整壓縮算法。

• ioConfig

  消費者的配置。對於kafak index service 就是kafka 消費者一個配置。

  下面的實例，配置了kafka的topic,啓動的任務數量，任務執行的時間，kafka的地址。

  completionTimeout：這個值將發佈任務聲明爲失敗並終止以前等待的時間。若是設置得過低，您的任務可能永遠不會發布。任務的發佈時間大約在taskDuration過去以後開始。默認是30M,爲防止任務未發佈，調整爲與任務時間一致（PT3600S）

"ioConfig": {        

"topic": "com.test",       

 "replicas": 1,         "taskCount": 1,     

   "taskDuration": "PT3600S",     

   "consumerProperties": {        

   "bootstrap.servers": "10.0.0.1:9096,10.0.0.1:9096"    

   },     

   "completionTimeout": "PT3600S"  

 }

提交supervisor

提交至overlord節點。

新版中出現界面配置

第一種，根據界面的配置嚮導來加載kafka數據

訪問：8888  端口

一直按照嚮導配置，就能夠自動生成supervisor的配置 很方便。

第二種，經過頁面 提供的Submit supervisor提交 相應的json文件

總結

簡單介紹了下supervisor  重點配置的具體含義，因爲篇幅問題，詳細的配置還須要去官網文檔中查看。本文的目的就是經過我的使用 kafka index service時一些新得，幫助新手能快速跑通第一個druid實例。

文章若有幫助，請關注微信公共號。