黑马程序员技术交流社区

标题: 【上海校区】----druid安装及入门-- [打印本页]

作者: xiaozuoquan 时间: 2020-1-3 14:45
标题: 【上海校区】----druid安装及入门--
本帖最后由 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使用的MapReduce批处理框架，数据何时能够查询没有性能保证
随机IO问题
- HDFS以集群硬盘作为存储资源池的分布式文件系统
- 在海量数据的处理过程中，会引起大量的读写操作，随机IO是高并发场景下的性能瓶颈
数据可视化问题
- HDFS对于数据分析以及数据的即席查询，HDFS并不是最优的选择

传统的Hadoop大数据处理架构更倾向于一种“后台批处理的数据仓库系统”，其作为海量历史数据保存、冷数据分析，确实是一个优秀的通用解决方案，但

无法保证高并发环境下海量数据的查询分析性能
无法实现海量实时数据的查询分析与可视化

4.2.2 Druid的诞生

Druid是由一个名为 MetaMarket 的公司开发的
2011年，MetaMarket 开始研发自己的"轮子"Druid，将Druid定义为“开源、分布式、面向列式存储的实时分析数据存储系统”
要解决的"痛点"是
- 在高并发环境下，保证海量数据查询分析性能
- 同时提供海量实时数据的查询、分析与可视化功能

4.3 Druid是什么

Druid是面向海量数据的、用于实时查询与分析的OLAP存储系统。Druid的四大关键特性如下：

亚秒级的OLAP查询分析
- 采用了列式存储、倒排索引、位图索引等关键技术
在亚秒级别内完成海量数据的过滤、聚合以及多维分析等操作
实时流数据分析
- 传统分析型数据库采用的批量导入数据，进行分析的方式
- Druid提供了实时流数据分析，以及高效实时写入
实时数据在亚秒级内的可视化
丰富的数据分析功能
- Druid提供了友好的可视化界面
SQL查询语言
- REST查询接口
高可用性与高可拓展性
- Druid工作节点功能单一，不相互依赖
- Druid集群在管理、容错、灾备、扩容都很容易

阿里巴巴也曾创建过一个开源项目叫 Druid （简称阿里 Druid），它是一个数据库连接池项目。阿里 Druid 和我们要讨论的 Druid 没有任何关系，它们解决完全不同的问题

4.4 国内哪些公司在使用Druid

腾讯
- 腾讯企点采用Druid用于分析大量的用户行为，帮助提升客户价值
阿里巴巴
- 阿里搜索组使用Druid的实时分析功能用于获取用户交互行为
新浪微博
- 新浪广告团队使用Druid构建数据洞察系统的实时分析部分，每天处理数十亿的消息
小米
- Druid用于小米统计的后台数据收集和分析
- 也用于广告平台的数据分析
滴滴打车
- Druid是滴滴实时大数据处理的核心模块，用于滴滴实时监控系统，支持数百个关键业务指标
- 通过Druid，滴滴能够快速得到各种实时的数据洞察
优酷土豆
- Druid用于其广告的数据处理和分析

4.5.1 Druid vs. Elasticsearch

Druid在导入过程会对原始数据进行Rollup，而ES会保存原始数据
Druid专注于OLAP，针对数据导入以及快速聚合操作做了优化
Druid不支持全文检索

4.5.2 Druid vs. Key/Value Stores (HBase/Cassandra/OpenTSDB)

Druid采用列式存储，使用倒排和bitmap索引，可以做到快速扫描相应的列

4.5.3 Druid vs. Spark

Spark SQL的响应还不做到亚秒
Druid可以做到超低的响应时间，例如亚秒，而且高并发面向用户的应用。

4.5.4 Druid vs SQL-on-Hadoop (Impala/Drill/Spark SQL/Presto)

Driud查询速度更快
数据导入，Druid支持实时导入，SQL-on-Hadoop一般将数据存储在Hdfs上，Hdfs的写入速度有可能成为瓶颈
SQL支持，Druid也支持SQL，但Druid不支持Join操作

4.6.5 Druid vs. Kylin

Kylin不支持实时查询，Druid支持
Kylin支持表连接（Join），Druid不支持

5. 安装imply

5.1 Imply介绍

Imply也是Druid的核心团队开发的，它基于Apache Druid开发了一整套大数据分析解决方案
Imply基于Druid进行了一些组件开发，提供开源社区版本和商业版，简化了部署

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;

注意事项

MySQL版本必须使用5.5及以上版本（Druid和Pivot使用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 文件

修改zookeeper的配置

druid.zk.service.host=node01:2181,node02:2181,node03:2181

修改MySQL的配置

druid.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=123456

4、修改 conf/pivot/config.yaml 配置文件

修改mysql的配置

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.4

source /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.conf

4、node03节点（启动broker和router）

/export/servers/imply-3.0.4/bin/supervise -c /export/servers/imply-3.0.4/conf/supervise/query.conf

注意事项

如果希望imply运行在后台，在每个执行命令后面加 --daemonize，

5.7 访问WebUI

组件名

URL

brokerhttp://node03:8888
coordinator、overlordhttp://node01:8081/index.html
middleManager、historicalhttp://node01:8090/console.html

6. Druid入门案例

需求：

使用Druid 分析 2019年5月8日按照商品分类、商品区域的产品订单总额

要实现该入门案例：

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 摄取本地文件

需求：

将摄取服务器本地上的 ad_event.json 数据到Druid中

操作步骤：

1、在某一个服务器节点中创建 /root/druid/data2 文件夹

2、上传数据文件和摄取配置文件

将资料文件夹中的 druid测试数据源\广告点击数据中的 ad_event.json 上传到 /root/druid/data2目录中
将 /root/druid/data2 目录分发到每个服务器节点

3、使用postman提交本地批量索引任务

将index_ad_event.json文件中的内容拷贝到 postman 中
发送post请求到http://node01:8090/druid/indexer/v1/task

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中查询数据。

需求：

摄取HDFS上的wikiticker-2015-09-12-sampled.json文件到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 提交索引任务

将index_wikiticker-2015-9-12-sample.json文件中的内容拷贝到 postman 中
发送post请求到http://node1:8090/druid/indexer/v1/task

5、到 Druid控制台中执行SQL查询

SELECT *
FROM "wikiticker"
LIMIT 1

7.2 流式（实时）数据摄取
7.2.1 Kafka索引服务方式摄取

需求：

实时摄取Kafka中 metrics topic的数据到 Druid中

操作步骤：

1、启动 Kafka 集群

2、在Kafka集群上创建一个名为metrics的topic

bin/kafka-topics.sh --create --zookeeper node01:2181,node02:2181,node03: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

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 主体结构

摄取配置文件主要由以下几个部分组成：

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": {...}
}
}
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
}
}

欢迎光临黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2