入門
本教程講述如何通過kafka加載數(shù)據(jù)到Druid。
在本教程中歧沪，我們假設(shè)您已經(jīng)按照快速入門中所述下載了Druid和Tranquility歹撒，并將其在本機上運行。并且您不需要事先加載數(shù)據(jù)诊胞。
本教程會指導(dǎo)如何通過kafka向Druid加載數(shù)據(jù)暖夭，但Druid還支持多種批量和流式加載數(shù)據(jù)的方法∧旃拢可以通過 Loading files and Loading streams頁面來學(xué)習(xí)其它更多數(shù)據(jù)加載的方法迈着。包括 Hadoop、HTTP邪码、Storm裕菠、Samza、Spark Streaming以及自研的JVM應(yīng)用
啟動kafka
Apache Kafka是一個高吞吐量的消息中間件闭专，可以和Druid配合使用糕韧。本教程中使用的是Kafka 0.9.0.0，可以通過如下指令下載kafka：
curl -O http://www.us.apache.org/dist/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz
tar -xzf kafka_2.11-0.9.0.0.tgz
cd kafka_2.11-0.9.0.0
執(zhí)行如下指令啟動kafka broker：
./bin/kafka-server-start.sh config/server.properties
創(chuàng)建一個名稱為metrics的topic用來接收數(shù)據(jù)：
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic metrics
發(fā)送示例數(shù)據(jù)
下面就可以開始通過console producer向kafka對應(yīng)的topic發(fā)送數(shù)據(jù)了喻圃！
在Druid目錄下執(zhí)行如下指令：
bin/generate-example-metrics
在kafka目錄下執(zhí)行：
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic metrics
現(xiàn)在kafka-console-producer就開始等待數(shù)據(jù)的輸入了萤彩，復(fù)制剛生成的示例數(shù)據(jù)并粘貼到kafka-console-producer控制臺終端，回車確認(rèn)斧拍。當(dāng)然也可以復(fù)制更多數(shù)據(jù)到終端雀扶，或者CTRL-D退出。
現(xiàn)在就可以進(jìn)行數(shù)據(jù)查詢了肆汹，當(dāng)然也可以參考下文去加載自定義數(shù)據(jù)集愚墓。
數(shù)據(jù)查詢
數(shù)據(jù)發(fā)送完成后就可以進(jìn)行數(shù)據(jù)查詢了，使用方法詳見 supported query methods.
加載自定義數(shù)據(jù)
目前為止昂勉，我們已經(jīng)按照Druid發(fā)布版本中的數(shù)據(jù)提取規(guī)范浪册，將數(shù)據(jù)從kafka加載到了Druid。每一個數(shù)據(jù)提取規(guī)范都是為了特定的數(shù)據(jù)集設(shè)計的岗照，也可以通過自定義提取規(guī)范來加載自定義數(shù)據(jù)村象。
自定義數(shù)據(jù)提取規(guī)范笆环，可以按需修改conf-quickstart/tranquility/kafka.json配置文件
- dataSchema，使用的數(shù)據(jù)集名稱
- timestampSpec厚者，哪個是時間字段
- dimensionsSpec躁劣，哪些能作為維度字段
- metricsSpec，哪些能作為度量進(jìn)行計算
{
"dataSources" : {
"metrics-kafka" : {
"spec" : {
"dataSchema" : {
"dataSource" : "metrics-kafka",
"parser" : {
"type" : "string",
"parseSpec" : {
"timestampSpec" : {
"column" : "timestamp",
"format" : "auto"
},
"dimensionsSpec" : {
"dimensions" : [],
"dimensionExclusions" : [
"timestamp",
"value"
]
},
"format" : "json"
}
},
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "hour",
"queryGranularity" : "none"
},
"metricsSpec" : [
{
"type" : "count",
"name" : "count"
},
{
"name" : "value_sum",
"type" : "doubleSum",
"fieldName" : "value"
},
{
"fieldName" : "value",
"name" : "value_min",
"type" : "doubleMin"
},
{
"type" : "doubleMax",
"name" : "value_max",
"fieldName" : "value"
}
]
},
"ioConfig" : {
"type" : "realtime"
},
"tuningConfig" : {
"type" : "realtime",
"maxRowsInMemory" : "100000",
"intermediatePersistPeriod" : "PT10M",
"windowPeriod" : "PT10M"
}
},
"properties" : {
"task.partitions" : "1",
"task.replicants" : "1",
"topicPattern" : "metrics"
}
}
},
"properties" : {
"zookeeper.connect" : "localhost",
"druid.discovery.curator.path" : "/druid/discovery",
"druid.selectors.indexing.serviceName" : "druid/overlord",
"commit.periodMillis" : "15000",
"consumer.numThreads" : "2",
"kafka.zookeeper.connect" : "localhost",
"kafka.group.id" : "tranquility-kafka"
}
}
下面使用網(wǎng)頁瀏覽為例并將輸入發(fā)送到pageviews的topic里库菲，示例數(shù)據(jù)如下：
{"time": "2000-01-01T00:00:00Z", "url": "/foo/bar", "user": "alice", "latencyMs": 32}
首先創(chuàng)建topic
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic pageviews
修改conf-quickstart/tranquility/kafka.json配置文件账忘，修改后：
{
"dataSources" : {
"metrics-kafka" : {
"spec" : {
"dataSchema" : {
"dataSource" : "pageviews-kafka",
"parser" : {
"type" : "string",
"parseSpec" : {
"timestampSpec" : {
"column" : "time",
"format" : "auto"
},
"dimensionsSpec" : {
"dimensions" : ["url", "user"],
"dimensionExclusions" : [
"timestamp",
"value"
]
},
"format" : "json"
}
},
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "hour",
"queryGranularity" : "none"
},
"metricsSpec" : [
{
"name": "views",
"type": "count"
},
{
"name": "latencyMs",
"type": "doubleSum",
"fieldName": "latencyMs"
}
]
},
"ioConfig" : {
"type" : "realtime"
},
"tuningConfig" : {
"type" : "realtime",
"maxRowsInMemory" : "100000",
"intermediatePersistPeriod" : "PT10M",
"windowPeriod" : "PT10M"
}
},
"properties" : {
"task.partitions" : "1",
"task.replicants" : "1",
"topicPattern" : "pageviews"
}
}
},
"properties" : {
"zookeeper.connect" : "localhost",
"druid.discovery.curator.path" : "/druid/discovery",
"druid.selectors.indexing.serviceName" : "druid/overlord",
"commit.periodMillis" : "15000",
"consumer.numThreads" : "2",
"kafka.zookeeper.connect" : "localhost",
"kafka.group.id" : "tranquility-kafka"
}
}
下面啟動Druid的kafka提取服務(wù)：
bin/tranquility kafka -configFile ../druid-0.9.2/conf-quickstart/tranquility/kafka.json
- 如果Tranquility或者kafka已經(jīng)啟動，可以停止并重新啟動熙宇。
最后將數(shù)據(jù)發(fā)送到kafka的topic鳖擒，以下面這些數(shù)據(jù)為例：
{"time": "2000-01-01T00:00:00Z", "url": "/foo/bar", "user": "alice", "latencyMs": 32}
{"time": "2000-01-01T00:00:00Z", "url": "/", "user": "bob", "latencyMs": 11}
{"time": "2000-01-01T00:00:00Z", "url": "/foo/bar", "user": "bob", "latencyMs": 45}
Druid流處理需要相對當(dāng)前（準(zhǔn)實時）的數(shù)據(jù)，相而言windowPeriod值控制的是更寬松的時間窗口（也就是流處理會檢查數(shù)據(jù)timestamp的值烫止，而時間窗口只關(guān)注數(shù)據(jù)接收的時間）败去。所以需要將2000-01-01T00:00:00Z轉(zhuǎn)換為ISO8601格式的當(dāng)前系統(tǒng)時間，你可以用以下命令轉(zhuǎn)換：
python -c 'import datetime; print(datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"))'
更新上述JSON中的時間戳烈拒，然后將這些消息復(fù)制并粘貼到此kafka-console-producer，然后按Enter鍵：
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic pageviews
就這樣广鳍，數(shù)據(jù)應(yīng)該已經(jīng)保存在Druid里了荆几，可以使用任何Druid支持的查詢方式查詢這些數(shù)據(jù)了。
進(jìn)一步閱讀
想了解更多流式數(shù)據(jù)加載赊时，可以查看streaming ingestion documentation
原文鏈接：http://druid.io/docs/0.9.2/tutorials/tutorial-kafka.html
