本帖最后由 xiaozuoquan 于 2020-1-3 14:53 编辑
4. Druid简介4.1 大数据分析平台架构分类数据分析的基础架构可以分为以下几类: 使用Hadoop/Spark进行分析 将Hadoop/Spark的结果导入 RDBMS 中提供数据分析 将结果注入到容量更大的 NoSQL中,解决数据分析的存储瓶颈,例如:HBase 将数据源进行流式处理,对接流式计算框架,例如:Flink、Spark Streaming,结果保存到 RDBMS、NoSQL中 将数据源进行流式处理,对接分析数据库,例如:Druid
4.2 为什么会有Druid
4.2.1 基于Hadoop大数据平台的问题
基于 Hadoop 的大数据平台,有如下一些问题:
传统的Hadoop大数据处理架构更倾向于一种“后台批处理的数据仓库系统”,其作为海量历史数据保存、冷数据分析,确实是一个优秀的通用解决方案,但 无法保证高并发环境下海量数据的查询分析性能 无法实现海量实时数据的查询分析与可视化
4.2.2 Druid的诞生
4.3 Druid是什么
Druid是面向海量数据的、用于实时查询与分析的OLAP存储系统。Druid的四大关键特性如下:
阿里巴巴也曾创建过一个开源项目叫 Druid (简称阿里 Druid),它是一个数据库连接池项目。阿里 Druid 和 我们要讨论的 Druid 没有任何关系,它们解决完全不同的问题
4.4 国内哪些公司在使用Druid
腾讯 阿里巴巴 新浪微博 小米 Druid用于小米统计的后台数据收集和分析 也用于广告平台的数据分析
滴滴打车 优酷土豆
4.5.1 Druid vs. Elasticsearch
4.5.2 Druid vs. Key/Value Stores (HBase/Cassandra/OpenTSDB)4.5.3 Druid vs. Spark4.5.4 Druid vs SQL-on-Hadoop (Impala/Drill/Spark SQL/Presto)4.6.5 Druid vs. Kylin5. 安装imply
5.1 Imply介绍
5.2 集群规划
主机名称 IP地址 角色 数据库
node01192.168.88.120zk、kafka、druid(overlord、coordinator)MySQL
node02192.168.88.121zk、kafka、druid(middleManager、historical)
node03192.168.88.122zk、kafka、druid(broker、router)
5.3 下载imply
Imply-3.0.4 基于 apache-druid-0.15.0-Incubating
1、下载imply cd /export/softwares/
wget https://static.imply.io/release/imply-3.0.4.tar.gz
2、直接使用资料 imply安装包jps' 将该 imply安装包\imply-3.0.4.tar.gz 安装包上传到 /exports/softwares
5.4 解压imply-3.0.4
tar -xvzf imply-3.0.4.tar.gz -C ../servers
cd ../servers/imply-3.0.4
5.5 配置imply-3.0.4
5.5.1 mysql中创建imply相关的数据库
CREATE DATABASE `druid` DEFAULT CHARACTER SET utf8;
CREATE DATABASE `pivot` DEFAULT CHARACTER SET utf8;
注意事项
5.5.2 修改并上传配置文件
1、将 imply 安装目录的 conf 目录重命名为 conf.bak mv conf conf.bak
1、上传 imply配置文件\conf.zip 到 imply-3.0.4 安装目录 2、解压缩 unzip conf.zip 3、修改 conf/druid/_common/common.runtime.properties 文件 druid.zk.service.host=node01:2181,node02:2181,node03:2181druid.metadata.storage.type=mysql
druid.metadata.storage.connector.connectURI=jdbc:mysql://node01:3306/druid
druid.metadata.storage.connector.user=root
druid.metadata.storage.connector.password=1234564、修改 conf/pivot/config.yaml 配置文件 stateStore:
type: mysql
location: mysql
connection: 'mysql://root:123456@node01:3306/pivot'
5.5.3 将配置好的 imply 分发到不同节点
scp -r imply-3.0.4/ node02:$PWD
scp -r imply-3.0.4/ node03:$PWD
5.5.4 配置环境变量
在每台服务器上配置DRUID_HOME环境变量
# DRUID
export DRUID_HOME=/export/servers/imply-3.0.4source /etc/profile 重新加载环境变量
5.6 启动 imply 集群
1、启动zk集群 2、node01节点(使用外部zk而不使用imply自带zk启动overlord和coordinator) # 使用外部zk而不使用imply自带zk启动overlord和coordinator
/export/servers/imply-3.0.4/bin/supervise -c /export/servers/imply-3.0.4/conf/supervise/master-no-zk.conf 3、node02节点(启动historical和middlemanager) /export/servers/imply-3.0.4/bin/supervise -c /export/servers/imply-3.0.4/conf/supervise/data.conf4、node03节点(启动broker和router) /export/servers/imply-3.0.4/bin/supervise -c /export/servers/imply-3.0.4/conf/supervise/query.conf
注意事项
5.7 访问WebUI组件名 URL
brokerhttp://node03:8888
coordinator、overlordhttp://node01:8081/index.html
middleManager、historicalhttp://node01:8090/console.html
6. Druid入门案例需求:
要实现该入门案例: 1、上传测试数据到每个Linux服务器 在一台Druid服务器中创建 /root/druid/data1/ 目录 mkdir -p /root/druid/data1 将druid测试数据源\商品订单数据\order.json到服务器的 /root/druid/data1/ 目录中 将 /root/druid/data1 分发到每一台服务器
2、摄取数据到Druid中 2.1 打开postman,请求地址设置为 http://node01:8090/druid/indexer/v1/task 2.2 请求方式选择为POST 2.3 body > raw > JSON(application/json) 2.4 将 资料中的index_order.json文件 粘贴到 postman中 2.5 发送请求
3、执行 SQL 查询 3.1 打开 Druid 控制台 http://node03:8888 3.2 打开 Query 选项卡,执行以下SQL实现 按照商品分类、商品区域的产品订单总额
-- 分析2019年5月8日,按照商品分类、商品区域的产品订单总额
SELECT
category,
areaName,
SUM(money) AS total_money,
SUM("count") AS total_count
FROM "demo_order"
WHERE TIME_FORMAT("__time", 'yyyyMMdd') = '20190508'
GROUP BY category, areaName
7. Druid数据摄取Druid支持流式和批量两种方式的数据摄入,针对不同类型的数据,Druid将外部数据源分为两种形式:
7.1 批量(离线)数据摄取
流式数据可以通过两种方式来摄入:
7.1.1 摄取本地文件需求:
操作步骤: 1、在某一个服务器节点中创建 /root/druid/data2 文件夹 2、上传数据文件和摄取配置文件 3、使用postman提交本地批量索引任务 4、可以在Overlord(http://node01:8090/console.html) 中查看到任务信息
6、在 http://node3:8888中测试查询数据 SELECT
*
FROM "ad_event"
LIMIT 1
7.1.2 摄取HDFS文件
Druid支持加载HDFS上的数据。它会使用 HadoopDruidIndexer 加载批量数据,将数据生成 segments 文件,存放在HDFS上,再从HDFS下载 segments 文件到本地。然后遍可从Druid中查询数据。
需求: 操作步骤: 1、启动HDFS集群、YARN集群 2、上传 druid测试数据源\维基百科访问日志数据到任意服务器 /root/druid/data3 目录,再将 wikiticker-2015-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 提交索引任务 5、到 Druid控制台中执行SQL查询 SELECT *
FROM "wikiticker"
LIMIT 1
7.2 流式(实时)数据摄取
7.2.1 Kafka索引服务方式摄取需求:
操作步骤: 1、启动 Kafka 集群 2、在Kafka集群上创建一个名为metrics的topic bin/kafka-topics.sh --create --zookeeper node01:2181,node02:2181,node03:2181, --partitions 1 --replication-factor 1 --topic metrics3、定义摄取配置文件 4、打开postman提交索引任务
4、在Kafka集群上开启一个控制台producer bin/kafka-console-producer.sh --broker-list node01:9092,node02:2181,node03:2181 --topic metrics
5、在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}
6、在 Druid Console中执行以下SQL查询 SELECT *
from "metrics-kafka"
LIMIT 1
7.3 摄取配置文件结构说明
7.3.1 主体结构
摄取配置文件主要由以下几个部分组成:
{
// ① 文件上传方式
// 1.1 index - 上传本地文件
// 1.2 index_hadoop - 上传HDFS文件
// 1.3 kafka - 拉取Kafka流数据
"type": "index",
"spec": {
// ② 数据解析模式
"dataSchema": {...},
// ③ 摄取数据源
"ioConfig": {...},
// ④ 摄取过程优化配置
"tuningConfig": {...}
}
}
7.3.2 数据解析模式
数据解析模式,主要为针对数据文件,定义了一系列规则: 获取时间戳属性 维度属性 度量属性 定义如何进行指标计算 配置粒度规则
// ② 数据摄取模式
"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": "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"
]
},
}
7.3.3 数据源配置
数据源配置主要指定: "ioConfig": {
"type": "index",
"inputSpec": {
// 3.1 本地文件 local/ HDFS使用 hadoop
"type": "local",
// 3.2 路径
"baseDir": "/root/data/",
// 3.3 只过滤出来哪个文件
"filter": "ad_event.json"
}
}
7.3.4 优化配置
通常在优化配置中可以指定一些优化选项 "tuningConfig": {
"type": "index",
// 4.1 分区类型
"partitionsSpec": {
"type": "hashed",
// 4.2 每个分区的目标行数(这里配置每个分区500W行)
"targetPartitionSize": 5000000
}
}
|