文章目录
一、数据格式
二、配置
2.1 DataSchema
2.1.1 parser
2.1.2 metricsSpec
2.1.3 GranularitySpec
2.2 ioConfig
2.3 tuningConfig
三、从 Hadoop 加载数据
3.1 加载数据
3.2 查询数据
四、从 kafka 加载数据
4.1 准备kafka
4.2 启动索引服务
4.3 加载历史数据
4.4 加载实时数据
4.5 加载自定义kafka 主题数据
一、数据格式
- 摄入规范化数据：JSON、CSV、TSV
- 自定义格式
- 其他格式
二、配置
主要是摄入的规则 Ingestion Spec
Ingestion Spec（数据格式描述）是Druid对要索引数据的格式以及如何索引该数据格式的一个统一描述，它是一个JSON文件，一般由三部分组成。
{ "dataSchema" : {...}, "ioConfig" : {...}, "tuningConfig" : {...} }
2.1 DataSchema
第一部分的dataSchema描述了数据的格式，如何解析该数据，典型结构如下。
{ "dataSource": <name_of_dataSource>, "parser": { "type": <>, "parseSpec": { "format": <>, "timestampSpec": {}, "dimensionsSpec": {} } }, "metricsSpec": {}, "granularitySpec": {} }
2.1.1 parser
parser部分决定了数据如何被正确地解析，metricsSpec定义了数据如何被聚集计算，granularitySpec定义了数据分片的粒度、查询的粒度。
对于parser，type有两个选项：string和hadoopString，后者用于Hadoop索引的 job。parseSpec是数据格式解析的具体定义。
（1）string parser
parseSpec 两个功能：
String Parser 用parseSpec 判定将要处理rows 的数据格式（ JSON, CSV, TSV）
所有的Parsers 用parseSpec 判定将要处理rows 的timestamp 和dimensionsAll
JSON ParseSpec
CSV ParseSpec
TSV ParseSpec
对于不同的数据格式，可能还有额外的parseSpec选项。
TimestampSpec
DimensionsSpec
2.1.2 metricsSpec
metricsSpec是一个JSON对象数组，定义了一些聚合器（aggregators）。聚合器通常有如下的结构。
{ "type": <type>, "name": <output_name>, "fieldName": <metric_name> }
2.1.3 GranularitySpec
聚合支持两种聚合方式：uniform和arbitrary，前者以一个固定的时间间隔聚合数据，后者尽量保证每个segments大小一致，时间间隔是不固定的。目前uniform是默认选项。
"dataSchema" : { "dataSource" : "wikipedia", "parser" : { "type" : "string", "parseSpec" : { "format" : "json", "dimensionsSpec" : { "dimensions" : [ "channel", "cityName", "comment", "countryIsoCode", "countryName", "isAnonymous", "isMinor", "isNew", "isRobot", "isUnpatrolled", "metroCode", "namespace", "page", "regionIsoCode", "regionName", "user", { "name" : "commentLength", "type" : "long" }, { "name" : "deltaBucket", "type" : "long" }, "flags", "diffUrl", { "name": "added", "type": "long" }, { "name": "deleted", "type": "long" }, { "name": "delta", "type": "long" } ] }, "timestampSpec": { "column": "timestamp", "format": "iso" } } }, "metricsSpec" : [], "granularitySpec" : { "type" : "uniform", "segmentGranularity" : "day", "queryGranularity" : "none", "intervals" : ["2016-06-27/2016-06-28"], "rollup" : false } }
2.2 ioConfig
ioConfig 指明了真正具体的数据源
不同的firehose 的格式不太一致，以kafka 为例：
{ firehose : { consumerProps : { auto.commit.enable : false auto.offset.reset : largest fetch.message.max.bytes : 1048586 group.id : druid-example zookeeper.connect : localhost:2181 zookeeper.connect.timeout.ms : 15000 zookeeper.session.timeout.ms : 15000 zookeeper.sync.time.ms : 5000 }, feed : wikipedia type : kafka-0.8 } }
ioConfig 的案例：
"ioConfig" : { "type" : "index", "firehose" : { "type" : "local", "baseDir" : "quickstart/", "filter" : "wikipedia-2016-06-27-sampled.json" }, "appendToExisting" : false }
2.3 tuningConfig
tuningConfig 这部分配置是优化数据输入的过程
"tuningConfig" : { "type" : "index", "targetPartitionSize" : 5000000, "maxRowsInMemory" : 25000, "forceExtendableShardSpecs" : true }
