Druid--数据摄取

Druid数据摄取分类

• Druid支持流式和批量两种方式的数据摄入，针对不同类型的数据，Druid将外部数据源分为两种形式：
  • 流式数据源
    • 指的是持续不断地生产数据的数据源。例如：消息队列、日志、文件等
  • 静态数据源
    • 指的是数据已经生产完毕，不会有新数据产生的数据源。例如：文件系统的文件

批量（离线）数据摄取

• 批量数据可以通过两种方式来摄入：

摄取本地文件

摄取HDFS文件

• Druid支持加载HDFS上的数据。它会使用 HadoopDruidIndexer 加载批量数据，将数据生成 segments 文件，存放在HDFS上，再从HDFS下载 segments 文件到本地。然后便可从Druid中查询数据。
• 需求：摄取HDFS上的wikiticker-2019-09-12-sampled.json文件到Druid中
• 操作步骤：
• 1、启动HDFS集群、YARN集群
• 2、上传 druid测试数据源\维基百科访问日志数据到任意服务器 /root/druid/data3 目录，再将 wikiticker-2019-09-12-sampled.json 文件上传到HDFS

hadoop fs -put wikiticker-2015-09-12-sampled.json /

• 3、修改 index_wikiticker-2015-9-12-sample.json 文件中配置 HDFS 的地址
• 4、使用 postman 提交索引任务
  • 将index_wikiticker-2019-9-12-sample.json文件中的内容拷贝到 postman 中
  • 发送post请求到http://node1:8090/druid/indexer/v1/task
• 5、到 Druid控制台中执行SQL查询

SELECT *
FROM "wikiticker"
LIMIT 1

流式（实时）数据摄取

Kafka索引服务方式摄取

• 需求：实时摄取Kafka中 metrics topic的数据到 Druid中
• 操作步骤：
• 1、启动 Kafka 集群
• 2、在Kafka集群上创建一个名为metrics的topic

bin/kafka-topics.sh --create --zookeeper node1:2181,node2:2181,node3:2181, --partitions 1 --replication-factor 1 --topic metrics

• 3、定义摄取配置文件

  • 修改 druid测试数据源\kafka实时摄取数据中的 index-metrics-kafka.json 文件中的kafka服务器地址

• 4、打开postman提交索引任务

  • 将 index-metrics-kafka.json 文件中的内容拷贝到 postman 中
  • 发送post请求到http://node1:8090/druid/indexer/v1/supervisor

• 在Overlord中可以看到
  

• 5、在Kafka集群上开启一个控制台producer

/export/servers/kafka_2.11-1.0.0/bin/kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic metrics

• 6、在Kafka producer控制台中粘贴如下数据

{"time":"2019-07-23T17:57:58Z","url":"/foo/bar","user":"alice","latencyMs":32}
{"time":"2019-07-23T17:57:59Z","url":"/","user":"bob","latencyMs":11}
{"time":"2019-07-23T17:58:00Z","url": "/foo/bar","user":"bob","latencyMs":45}

• 7、在 Druid Console中执行以下SQL查询

SELECT *
from "metrics-kafka"
LIMIT 1

摄取配置文件结构说明

主体结构

• 摄取配置文件主要由以下几个部分组成：
  • type：文件上传方式（index、index_hadoop、kafka）
  • spec
    • dataSchema：数据解析模式
    • ioConfig：数据源
    • turningConfig：优化配置（分区规则、分区大小）

{// ① 文件上传方式// 1.1 index        - 上传本地文件// 1.2 index_hadoop - 上传HDFS文件// 1.3 kafka        - 拉取Kafka流数据"type": "index","spec": {// ② 数据解析模式"dataSchema": {...},// ③ 摄取数据源"ioConfig": {...},// ④ 摄取过程优化配置"tuningConfig": {...}}
}

数据解析模式

• 数据解析模式，主要为针对数据文件，定义了一系列规则：
  • 取时间戳属性
  • 维度属性
  • 度量属性
  • 定义如何进行指标计算
  • 配置粒度规则

// ② 数据摄取模式
"dataSchema": {// 2.1 数据源（表）"dataSource": "ad_event_local",// 2.2 解析器"parser": {// 2.2.1 解析字符串文本"type": "String","parseSpec": {// 2.2.1.1 字符串文本格式为JSON"format": "json",// 2.2.1.2 指定维度列名，维度与时间一致，导入时聚合"dimensionsSpec": {"dimensions": ["city","platform"]},// 2.2.1.3 指定时间戳的列，以及时间戳格式化方式"timestampSpec": {"format": "auto","column": "timestamp"}}},// 2.3 指标计算规则"metricsSpec": [{//name表示列名"name": "count","type": "count"},{// 2.3.1 聚合计算后指标的列名"name": "click",// 2.3.2 聚合函数：count、longSum、doubleSum、longMin、doubleMin、doubleMax"type": "longSum","fieldName": "click"}]// 2.4 粒度规则"granularitySpec": {"type": "uniform",// 2.4.1 按天来生成 segment （每天生成一个segment）"segmentGranularity": "day",// 2.4.2 查询的最小粒度（最小粒度为小时）"queryGranularity": "hour",// 2.4.3 加载原始数据的时间范围，批量数据导入需要设置/流式导入无需设置"intervals": ["2018-12-01/2018-12-03"]},}

数据源配置

• 数据源配置主要指定：
  • 要加载数据的类型
  • 从哪儿加载数据

"ioConfig": {"type": "index","inputSpec": {// 3.1 本地文件 local/ HDFS使用 hadoop"type": "local",// 3.2 路径"baseDir": "/root/data/",// 3.3 只过滤出来哪个文件"filter": "ad_event.json"}
}

优化配置

• 通常在优化配置中可以指定一些优化选项

"tuningConfig": {"type": "index",// 4.1 分区类型"partitionsSpec": {"type": "hashed",// 4.2 每个分区的目标行数（这里配置每个分区500W行）"targetPartitionSize": 5000000}
}

了解Druid WebUI生成 spec