0836-Apache Druid on HDP
0836-Apache Druid on HDP
作者：卢其敏
1.Apache Druid简介
Apache Druid是一个分布式的、面向列的、实时分析数据库，旨在快速获取大量数据并将其编入索引，并对大型数据集进行快速的切片和切分分析（“OLAP查询），常用于实时摄取、快速查询和对时间依赖性很高的数据库用户。因此，Druid可以为可视化的分析应用程序提供强力的数据源支持，或用作需要快速聚合的高并发API的后端。Druid最适合面向事件的数据。
Apache Druid通常位于存储或处理层与最终用户之间，并充当查询层以服务于分析工作负载。
常见应用领域包括：点击流分析、网络遥测分析、服务器指标存储、供应链分析、应用程序性能指标、数字营销、广告分析、商业智能BI / OLAP等。
Apache Druid的核心架构结合了数据仓库、时间序列数据库和日志搜索系统的思想，包括以下主要功能：
- 列式存储格式
- 可扩展的分布式系统
- 大规模并行处理
- 实时或批量加载数据
- 自我修复、自我平衡、易于操作
- 云原生的容错架构，不会丢失数据
- 用于快速过滤的索引
- 基于时间的分区
- 近似算法
- 加载数据时自动汇总
2.Apache Druid架构
Apache Druid具有多进程，分布式架构，旨在实现云友好且易于操作。每种Druid进程类型都可以独立配置和扩展，从而为您的集群提供最大的灵活性。这种设计还提高了容错能力：一个组件的故障不会立即影响其他组件。
2.1 服务器类型
我们一般将Druid的服务器分为三种类型：主服务器（Master Server），查询服务器（Query Server）和数据服务器（Data Server）。
2.1.1 Master Server
Master Server管理数据的加载和可用性：它负责启动新的加载作业，并协调下述“Data Server”上数据的可用性。包含两个处理进程：Coordinator 和 Overlord。
- Coordinator进程监视数据服务器上的Historical进程，它主要负责Segment的管理和分配。更具体地说，Druid Coordinator进程与Historical进程进行通信，以基于配置加载或删除Segment。Druid Coordinator负责加载新的Segment、删除过时的Segment、管理Segment的复制以及平衡Segment的负载，确保Segment在所有的Historical记录之间保持平衡。
- Overlord进程监视数据服务器上的MiddleManager进程，并且是将数据加载到Druid中的控制器。它负责接受任务、协调任务分配、围绕任务创建锁以及将状态返回给调用方，并将加载任务分配给MiddleManager，并负责协调Segment的发布。可以将Overlord配置为以两种模式之一运行：本地模式或远程模式。
- 在本地模式下，Overlord还负责创建用于执行任务的Peon。在本地模式下运行Overlord时，还必须提供所有MiddleManager和Peon配置。本地模式通常用于简单的工作流程。
- 在远程模式下，Overlord和MiddleManager在单独的进程中运行，可以在不同的服务器上运行它们。如果打算将indexing服务用作整个Druid集群的索引服务，则建议使用此模式。
2.1.2 Query
Query Server提供用户和客户端应用程序与之交互的端点，将查询路由到数据服务器或其他查询服务器。包含两个处理进程：Broker和Router。
- Broker进程从外部客户端接收查询，并将这些查询转发到数据服务器。当Broker从这些子查询中接收到结果时，它们会合并这些结果并将其返回给调用方。最终用户通常查询Broker，而不是直接查询数据服务器上的Historicals或MiddleManagers进程。
- Router进程是一个可选的进程，它可以在Druid Broker、Overlord和Coordinator之前提供统一的API网关。
Router还运行Druid控制台，Druid控制台是用于数据源、段、任务、数据处理（Historical和MiddleManager）以及Coordinator动态配置的管理UI。还可以在控制台中运行SQL和Native Druid查询。
2.1.3 Data Server
Data Server：执行数据加载作业并存储可查询的数据。包含两个进程：Historical 和 MiddleManager。
- Historical是存储和查询“历史”数据的主要进程，它从Deep Storage中下载Segment，并响应有关这些Segment的查询。不接受写操作。
- MiddleManager是将新数据加载到群集中的进程，负责从外部数据源读取数据并发布至新的Druid Segment。
- Peon进程是由MiddleManager产生的任务执行引擎，每个Peon运行一个单独的JVM，并负责执行一个任务。Peons始终与生成它们的MiddleManager在同一主机上运行。
2.2 外部依赖
除了内置的进程类型外，Druid还需要三个外部依赖项，可以利用现有的现有基础结构：Deep Storage、Metadata Storage、Zookeeper。
2.2.1 Deep Storage
Deep Storage是存储Segment的地方，Apache Druid本身不提供存储机制。这种Deep Storage的基础架构定义了数据的持久性级别，只要Druid进程可以看到该存储基础架构并能够获取存储在其上的Segment，那么无论丢失多少个Druid节点，数据都不会丢失。如果Segment从该存储层消失，则将丢失这些Segment表示的所有数据。
支持本地文件系统、HDFS和S3等，由属性druid.storage.type和druid.storage.storageDirectory等属性指定。
2.2.2 Metadata Storage
Metadata Storage是Apache Druid的外部依赖项，Apache Druid使用它来存储有关系统的各种元数据，而不是存储实际数据。
支持Derby、MySQL、PostgreSQL，由属性druid.metadata.storage.type等属性指定。
2.2.3 Zookeeper
Apache Druid使用Apache ZooKeeper（ZK）来管理当前集群状态，包含：
- Coordinator的Leader选举
- Historical中Segment的“发布”协议
- Coordinator和Historical之间Segment的加载/删除协议
- Overlord的Leader选举
- Overlord和MiddleManager的任务管理
2.3 存储设计
Druid的数据存储在“datasources”中，类似于传统RDBMS中的“table”。每个datasource都按时间分区，并且可以选择按其他属性进一步分区。每个时间范围都称为“chunk”（如果按天划分，则为一天）。在一个chunk内，数据被划分为一个或多个“segment”。每个segment都是单个文件，通常包含多达几百万行的数据。
一个datasource可能具有从几个segment到数十万甚至数百万个segment，每个segment都是从在MiddleManager上创建开始的，Segment的构建旨在生成紧凑且支持快速查询的数据文件，包括以下步骤：
- 转换为列格式
- 使用位图索引编制索引
- 使用各种算法进行压缩
- 字符串列的ID存储最小化的字典编码
- 位图索引的位图压缩
- 所有列的类型感知压缩
Apache Druid将其索引存储在Segment文件中，该Segment文件按时间进行分区。在基本设置中，将为每个时间间隔创建一个分段文件，其中该时间间隔可在granularitySpec的segmentGranularity参数中配置。为了使Druid在繁重的查询负载下正常运行，建议Segment文件的大小在300MB-700MB范围内。如果Segment文件大于此范围，可以更改时间间隔的粒度或者对数据进行分区，并在partitionsSpec中调整targetPartitionSize（一般建议最小为500万行）。
在Apache Druid中，一般有三种基本列的类型：时间戳列、维度列和指标列，如图所示：
时间戳和指标列，都是由LZ4压缩的整数或浮点值的数组。
维度列由于支持筛选和分组操作，一般需要以下三个数据结构：
- 将维度的值映射到整数ID的字典
- 使用上述字典编码的维度的值的列表
- 指示哪些行包含维度值的BITMAP
例如：
1: Dictionary that encodes column values
{
"Justin Bieber": 0,
"Ke$ha": 1
}
2: Column data
[0,
0,
1,
1]
3: Bitmaps - one for each unique value of the column
value="Justin Bieber": [1,1,0,0]
value="Ke$ha": [0,0,1,1]
3.在HDP上安装Apache Druid
环境 | 版本 |
|---|---|
操作系统 | RHEL-7.6 |
数据库 | MySQL-5.7 |
HDP | 3.1.4 |
3.1 准备数据库
创建数据库，并授权（Druid 数据库需要使用utf8编码）：
mysql> CREATE DATABASE druid DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci;
mysql> GRANT ALL PRIVILEGES ON efm.* TO efm@‘%’ IDENTIFIED BY ‘Cloudera4u’;
mysql> FLUSH PRIVILEGES;
加载MySQL的JDBC驱动：
ambari-server setup --jdbc-db=mysql --jdbc-driver=/usr/share/java/mysql-connector-java.jar
ambari-server restart
3.2 安装Druid
打开Ambari WebUI上的添加服务向导，勾选 Druid：
为Master Server和Query Server分配主机节点：
为 Data Server分配主机节点：
根据提示填入Metadata Storage数据库连接信息：
安装完成后，可以看到Druid的服务汇总页面：
在Druid服务汇总页面右侧，可以看到Quick Links下提供了两个WEB控制台
- Druid Coordinator Console，用于显示集群信息：
- Druid Overlord Console可用于查看挂起的任务、正在运行的任务、可用的工作程序以及最近创建和终止的任务：
4.导入数据
Apache Druid支持流式和批量加载数据两种方式，每种加载方法都支持其自己的源系统集。
- 批量加载：当从文件进行批量加载时，应使用一次性任务，并且支持三种类型：index_parallel（本地、可以并行）、index_hadoop（基于hadoop）、和index（本地、单线程）。
- 流式加载：最推荐、最流行的流式数据加载方法是直接从Kafka读取的Kafka索引服务。
无论使用哪种数据加载方式，都需要定制数据加载规范（JSON文件），主要由三个部分组成：
- dataSchema：定义数据源的名称、时间戳、维度、指标、转换和过滤器
- ioConfig：定义如何连接到数据源，以及如何解析数据
- tuningConfig：控制每种加载方法特有的各种参数
4.1 导入本地数据源
使用单线程批量加载的方式加载数据到Druid，
数据文件路径：/usr/hdp/current/druid-overlord/quickstart/wikiticker-2015-09-12-sampled.json.gz
4.1.1 定义规范
[root@hadoop47 ~]# cat index_local.json
{
"type" : "index",
"spec" : {
"dataSchema" : {
"dataSource" : "wikipedia_local",
"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": "added", "type": "long" },
{ "name": "deleted", "type": "long" },
{ "name": "delta", "type": "long" }
]
},
"timestampSpec": {
"column": "time",
"format": "iso"
}
}
},
"metricsSpec" : [],
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "day",
"queryGranularity" : "none",
"intervals" : ["2015-09-12/2015-09-13"],
"rollup" : false
}
},
"ioConfig" : {
"type" : "index",
"firehose" : {
"type" : "local",
"baseDir" : "/usr/hdp/current/druid-overlord/quickstart/",
"filter" : "wikiticker-2015-09-12-sampled.json.gz"
},
"appendToExisting" : false
},
"tuningConfig" : {
"type" : "index",
"maxRowsPerSegment" : 5000000,
"maxRowsInMemory" : 25000,
"forceExtendableShardSpecs" : true
}
}
}
4.1.2 加载数据
curl -X 'POST' -H 'Content-Type:application/json' -d @index_local.json http://hadoop46.luqimin.cn:8090/druid/indexer/v1/task
4.2 导入HDFS数据源
4.2.1 定义规范
{
"type" : "index_hadoop",
"spec" : {
"ioConfig" : {
"type" : "hadoop",
"inputSpec" : {
"type" : "static",
"paths" : "wikiticker-2015-09-12-sampled.json.gz"
}
},
"dataSchema" : {
"dataSource" : "wikiticker-hadoop",
"granularitySpec" : {
"type" : "uniform",
"segmentGranularity" : "day",
"queryGranularity" : "none",
"intervals" : ["2015-09-12/2015-09-13"]
},
"parser" : {
"type" : "hadoopyString",
"parseSpec" : {
"format" : "json",
"dimensionsSpec" : {
"dimensions" : [
"channel",
"cityName",
"comment",
"countryIsoCode",
"countryName",
"isAnonymous",
"isMinor",
"isNew",
"isRobot",
"isUnpatrolled",
"metroCode",
"namespace",
"page",
"regionIsoCode",
"regionName",
"user"
]
},
"timestampSpec" : {
"format" : "auto",
"column" : "time"
}
}
},
"metricsSpec" : [
{
"name" : "count",
"type" : "count"
},
{
"name" : "added",
"type" : "longSum",
"fieldName" : "added"
},
{
"name" : "deleted",
"type" : "longSum",
"fieldName" : "deleted"
},
{
"name" : "delta",
"type" : "longSum",
"fieldName" : "delta"
},
{
"name" : "user_unique",
"type" : "hyperUnique",
"fieldName" : "user"
}
]
},
"tuningConfig" : {
"type" : "hadoop",
"partitionsSpec" : {
"type" : "hashed",
"targetPartitionSize" : 5000000
},
"jobProperties" : {}
}
}
}
4.2.2 加载数据
# 上传数据文件到HDFS
su druid -l -c 'hdfs dfs -put wikiticker-2015-09-12-sampled.json.gz /user/druid/'
# 提交任务，该任务将提交至YARN运行
curl -X 'POST' -H 'Content-Type:application/json' -d @ wikiticker-index.json http://hadoop46.luqimin.cn:8090/druid/indexer/v1/task
4.3 导入Kafka数据源
打开Ambari中Druid的配置页面，修改Advanced druid-common中的属性druid.extensions.loadList，增加值：“druid-kafka-indexing-service”后，重启Druid服务。
4.3.1 定义规范
{
"type": "kafka",
"dataSchema": {
"dataSource": "wikipedia-kafka",
"parser": {
"type": "string",
"parseSpec": {
"format": "json",
"timestampSpec": {
"column": "time",
"format": "auto"
},
"dimensionsSpec": {
"dimensions": [
"channel",
"cityName",
"comment",
"countryIsoCode",
"countryName",
"isAnonymous",
"isMinor",
"isNew",
"isRobot",
"isUnpatrolled",
"metroCode",
"namespace",
"page",
"regionIsoCode",
"regionName",
"user",
{ "name": "added", "type": "long" },
{ "name": "deleted", "type": "long" },
{ "name": "delta", "type": "long" }
]
}
}
},
"metricsSpec" : [],
"granularitySpec": {
"type": "uniform",
"segmentGranularity": "DAY",
"queryGranularity": "NONE",
"rollup": false
}
},
"tuningConfig": {
"type": "kafka",
"reportParseExceptions": false,
"maxRowsInMemory": 1000
"maxRowsPerSegment": 5000000
},
"ioConfig": {
"topic": "wikipedia",
"replicas": 1,
"taskDuration": "PT10M",
"completionTimeout": "PT20M",
"consumerProperties": {
"bootstrap.servers": "hadoop45.luqimin.cn:6667,hadoop46.luqimin.cn:6667,hadoop47.luqimin.cn:6667"
}
}
}
4.3.2 提交任务
curl -X 'POST' -H 'Content-Type:application/json' -d @wikipedia-kafka-supervisor.json http://hadoop46.luqimin.cn:8090/druid/indexer/v1/supervisor
可以看到任务正在运行
向Kafka生产数据：
/usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list hadoop45.luqimin.cn:6667,hadoop47.luqimin.cn:6667,hadoop46.luqimin.cn:6667 --topic wikipedia < wikiticker-2015-09-12-sampled.json
这时可以立即查询Druid中的数据。
5.查询数据
5.1 Json over HTTP
5.1.1 定义规范
[root@hadoop47 ~]# cat wickiticker-top.json
{
"queryType" : "topN",
"dataSource" : "wikipedia_local",
"intervals" : ["2015-09-10/2015-09-14"],
"granularity" : "all",
"dimension" : "page",
"metric" : "count",
"threshold" : 10,
"aggregations" : [
{
"type" : "count",
"name" : "count"
}
]
}
5.1.2 提交查询任务
curl -X 'POST' -H 'Content-Type:application/json' -d @wickiticker-top.json http://hadoop45.luqimin.cn:8082/druid/v2?prett
返回结果：
[
{"timestamp":"2015-09-12T00:46:58.771Z",
"result":[
{"count":33,"page":"Wikipedia:Vandalismusmeldung"},
{"count":28,"page":"User:Cyde/List of candidates for speedy deletion/Subpage"},
{"count":27,"page":"Jeremy Corbyn"},
{"count":21,"page":"Wikipedia:Administrators' noticeboard/Incidents"},
{"count":20,"page":"Flavia Pennetta"},
{"count":18,"page":"Total Drama Presents: The Ridonculous Race"},
{"count":18,"page":"User talk:Dudeperson176123"},
{"count":18,"page":"Wikipédia:Le Bistro/12 septembre 2015"},
{"count":17,"page":"Wikipedia:In the news/Candidates"},
{"count":17,"page":"Wikipedia:Requests for page protection"}
]
}
]
5.2 SQL over HTTP
打开Ambari中Druid的配置页面，在Custom druid-common中增加属性druid.sql.enable = true，重启Druid服务。
5.2.1 定义查询
{"query": "select page, count(*) as c from \"wikipedia-kafka\" group by page order by c desc limit 5" }
5.2.2 提交查询
curl -X 'POST' -H 'Content-Type:application/json' -d @query.json http://hadoop45.luqimin.cn:8082/druid/v2/sql
返回结果：
[
{"page":"Wikipedia:Vandalismusmeldung","c":33},
{"page":"User:Cyde/List of candidates for speedy deletion/Subpage","c":28},
{"page":"Jeremy Corbyn","c":27},
{"page":"Wikipedia:Administrators' noticeboard/Incidents","c":21},
{"page":"Flavia Pennetta","c":20}
]
6.使用Druid加速Hive查询
可以使用Hive和Apache Druid的HDP集成对实时和历史数据执行交互式分析查询。可以发现现有的Druid数据源作为外部表，将批处理数据创建或摄取到Druid，使用Hive设置Druid-Kafka流式摄取，以及从Hive查询Druid数据源。
Hive与Druid的集成相当于在Druid上放置了一个SQL层。在Druid从Hive企业数据仓库（EDW）提取数据之后，可以使用Druid的交互式和亚秒级查询功能来加速对EDW中历史数据的查询。
6.1 配置
hive中跟druid相关的配置：（使用Ambari安装Druid时自动配置的Advanced hive-interactive-site）
hive.druid.bitmap.type=roaring
hive.druid.broker.address.default=hadoop45.luqimin.cn:8888
hive.druid.coordinator.address.default=hadoop46.luqimin.cn:8081
hive.druid.http.numConnection=20
hive.druid.http.read.timeout=PT10M
hive.druid.indexer.memory.rownum.max=75000
hive.druid.indexer.partition.size.max=1000000
hive.druid.indexer.segments.granularity=DAY
hive.druid.maxTries=5
hive.druid.metadata.base=druid
hive.druid.metadata.db.type=mysql
hive.druid.metadata.uri=jdbc:mysql://hadoop47.luqimin.cn:3306/druid?createDatabaseIfNotExist=true
hive.druid.metadata.username=druid
hive.druid.overlord.address.default=hadoop46.luqimin.cn:8090
hive.druid.passiveWaitTimeMs=30000
hive.druid.rollup=true
hive.druid.select.distribute=true
hive.druid.select.threshold=10000
hive.druid.sleep.time=PT10S
hive.druid.storage.storageDirectory=/apps/druid/warehouse
hive.druid.working.directory=/tmp/druid-indexing
Druid加载数据时，会进行自动汇总，临时关闭自动汇总请在beeline中设置：
set hive.druid.rollup=false
6.2 示例
样例数据
[root@hadoop47 ~]# head -n 1 wikiticker-2015-09-12-sampled.json
{"time":"2015-09-12T00:46:58.771Z","channel":"#en.wikipedia","cityName":null,"comment":"added project","countryIsoCode":null,"countryName":null,"isAnonymous":false,"isMinor":false,"isNew":false,"isRobot":false,"isUnpatrolled":false,"metroCode":null,"namespace":"Talk","page":"Talk:Oswald Tilghman","regionIsoCode":null,"regionName":null,"user":"GELongstreet","delta":36,"added":36,"deleted":0}
使用beeline连接Hive LLAP实例，将数据加载至Hive：
# 创建外部表wiki_json,加载Json数据文件
CREATE EXTERNAL TABLE wiki_json(json string)
row format delimited fields terminated by '\n'
stored as textfile
location '/tmp/json’;
# 创建内部表wiki
create table wiki(
`time` string,
`channel` string,
`cityName` string,
`comment` string,
`countryIsoCode` string,
`countryName` string,
`isAnonymous` string,
`isMinor` string,
`isNew` string,
`isRobot` string,
`isUnpatrolled` string,
`metroCode` string,
`namespace` string,
`page` string,
`regionIsoCode` string,
`regionName` string,
`user` string,
`delta` int,
`added` int,
`deleted` int
) ;
# 使用json_tuple函数获取json内部，并写入表wiki
insert overwrite table wiki select json_tuple(json,
'time',
'channel',
'cityName',
'comment',
'countryIsoCode',
'countryName',
'isAnonymou',
'isMinor',
'isNew',
'isRobot',
'isUnpatrolled',
'metroCode',
'namespace',
'page',
'regionIsoCode',
'regionName',
'user',
'delta',
'added',
'deleted') from wiki_json;
创建一个Druid表，与Hive表的字段对应：
CREATE external TABLE wiki_druid
STORED BY 'org.apache.hadoop.hive.druid.DruidStorageHandler'
TBLPROPERTIES (
"druid.segment.granularity" = "DAY",
"druid.query.granularity" = "none")
AS SELECT
cast(regexp_replace(`time`, "T|Z", " ") as timestamp) as `__time`,
cast(`channel` as string) `channel`,
cast(`cityname` as string) `cityname`,
cast(`comment` as string) `comment`,
cast(`countryisocode` as string) `countryisocode`,
cast(`countryname` as string) `countryname`,
cast(`isanonymous` as string) `isanonymous`,
cast(`isminor` as string) `isminor`,
cast(`isnew` as string) `isnew`,
cast(`isrobot` as string) `isrobot`,
cast(`isunpatrolled` as string) `isunpatrolled`,
cast(`metrocode` as string) `metrocode`,
cast(`namespace` as string) `namespace`,
cast(`page` as string) `page`,
cast(`regionisocode` as string) `regionisocode`,
cast(`user` as string) `user`,
cast(`delta` as int) `delta`,
cast(`added` as int) `added`,
cast(`deleted` as int) `deleted`
FROM wiki;
也可以创建一个Hive的物化视图，并将其存储在Druid中：
create materialized view wiki_view_druid
STORED BY 'org.apache.hadoop.hive.druid.DruidStorageHandler'
as select
cast(regexp_replace(`time`, "T|Z", " ") as timestamp) as `__time`,
`page`,
`user`,
`added`,
`delta`
from wiki;
执行查询
select page, count(*) as c from wiki_druid group by page order by c desc limit 5;
查看执行计划：
explain select page, count(*) as c from wiki_druid group by page order by c desc limit 5;
7.新版本UI
目前社区最新的Apache Druid稳定版本是0.17.0，除了功能增加和系统稳定性之外，还提供了全新的Web UI，如
- 可视化的数据加载页面
- 数据预览、过滤、转换、聚合等
- SQL执行界面
