A股上市公司传智教育(股票代码 003032)旗下技术交流社区北京昌平校区

 找回密码
 加入黑马

QQ登录

只需一步,快速开始

本帖最后由 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
    }
}



0 个回复

您需要登录后才可以回帖 登录 | 加入黑马