一、数据仓库概念
二、项目需求及架构设计
- 2.1 项目需求分析
- 2.2 项目框架
- - 2.2.1 技术选型
  - 2.2.2 系统数据流程设计
  - 2.2.3 框架版本选型
  - 2.2.4 服务器选型
  - 2.2.5 集群规模
  - 2.2.6 集群资源规划设计
三、数据生成模块
- 3.1 目标数据
- - 3.1.1 页面日志
  - 3.1.2 事件日志
  - 3.1.3 曝光日志
  - 3.1.4 启动日志
  - 3.1.5 错误日志
- 3.2数据埋点
- - 3.2.1 主流埋点方式
  - 3.2.2 埋点数据上报时机
  - 3.2.3 埋点数据日志结构
- 3.3 服务器和JDK准备
- 3.4 模拟数据
- - 3.4.1 使用说明
  - 3.4.2 集群日志生成脚本
四、数据采集模块
- 4.1 集群所有进程查看脚本
- 4.2 zookeeper安装
- - 4.2.1 安装ZK
  - 4.2.2 ZK集群启动停止脚本
- 4.3 Kafka安装
- - 4.3.1 Kafka集群安装
  - 4.3.2 Kafka集群启动停止脚本
  - 4.3.3 常用命令
  - 4.3.4 项目经验之Kafka机器数量计算
  - 4.3.5 项目经验之压力测试
  - 4.3.6 项目经验之Kafka分区数计算
- 4.4采集日志Flume
- - 4.4.1 日至采集Flume安装
  - 4.4.2 项目经验之Flume组件选型
  - 4.4.3 日志采集 Flume 配置
  - 4.4.4 Flume拦截器
  - 4.4.5 测试Flume-Kafka通道
  - 4.4.6 日至采集Flume启动停止脚本
- 4.5 消费Kafka数据Flume
- - 4.5.1 项目经验之Flume组件选型
  - - 4.5.1.1 FileChannel 和 MemoryChannel 区别
    - 4.5.1.2 FileChannel优化
    - 4.5.1.3 Sink：HDFS Sink
  - 4.5.2 Flume时间戳拦截器
  - 4.5.3 消费者Flume配置
  - 4.5.4 消费者Flume启动停止脚本
- 4.6 采集通道启动停止脚本
五、常见问题及解决方案
- 5.1 2NN页面不能显示完整信息

-----------------------------------------------------分隔符-----------------------------------------------------
数据仓库之电商数仓-- 1、用户行为数据采集==>
数据仓库之电商数仓-- 2、业务数据采集平台==>
数据仓库之电商数仓-- 3.1、电商数据仓库系统(DIM层、ODS层、DWD层)==>
数据仓库之电商数仓-- 3.2、电商数据仓库系统(DWS层)==>
数据仓库之电商数仓-- 3.3、电商数据仓库系统(DWT层)==>
数据仓库之电商数仓-- 3.4、电商数据仓库系统(ADS层)==>
数据仓库之电商数仓-- 4、可视化报表Superset==>
数据仓库之电商数仓-- 5、即席查询Kylin==>

一、数据仓库概念

数据仓库(Data Warehouse),是为企业制定决策，提供数据支持的。可以帮助企业改进业务流程、提高产品质量等。

数据仓库的输入数据通常包括：业务数据、用户行为数据和爬虫数据等。

业务数据：就是各行业在处理事务过程中产生的数据。如用户在电商网站中登录、下载、支付等过程中，需要和网站后台数据库进行增删改查交互，产生的数据就是业务数据。业务数据通常存储在MySQL、Oracle等数据库中。

二、项目需求及架构设计

2.1 项目需求分析

项目需求

用户行为数据采集平台搭建；
业务数据采集平台搭建；
数据仓库纬度建模；
分析设备、会员、商品、地区、活动等电商核心主题，统计的报表接近100个；
采用即席查询工具，随时进行指标分析；
对集群性能进行监控，发生异常需要报警；
元数据管理；
质量监控；
权限管理；

2.2 项目框架

2.2.1 技术选型

项目技术如何选型？
框架版本如何选型(Apache、CDH、HDP)
服务器使用物理机还是云主机？
如何确认集群规模？

⭐️：
技术选型主要考虑因素：数据量大小、业务需求、行业内经验、技术成熟度、开发维护成本、总成本预算。

数据采集传输：Flume, Kafka, Sqoop, Logstash, DataX;
数据存储：MySQL, HDFS, HBase, Redis, MongoDB;
数据计算：Hive, Tex, Spark, Flink, Storm;
数据查询：presto, Kylin, Impala, Druid, Clickouse, Doris;
数据可视化：Echarts, Superset, QuickBI, DataV;
任务调度：Azkaban, Oozie, DolphinScheduler, Airflow;
集群监控：Zabbix, Prometheus;
元数据管理：Atalas;
权限管理：Ranger, Sentry.

2.2.2 系统数据流程设计

2.2.3 框架版本选型

如何选择Apache/CDH/HDP版本？

1). Apache：运维麻烦，组件间兼容性需要自己调研，开源；
2). CDH：国内使用最多的版本，不开源；
3). HDP：开源，可进行二次开发，但没有CDH稳定，国内使用甚少。

云服务选择

1). 阿里云EMR、MaxCompute、DataWorks
2). 亚马逊云EMR
3). 腾讯云EMR
4). 华为云EMR

具体版本型号

Apache框架版本：

tips：
框架选型最好选择半年前的稳定版！

2.2.4 服务器选型

物理机：需专业运维人员；
云主机：若是选择阿里云，运维工作全由阿里云完成。

2.2.5 集群规模

如何确认集群规模？（假设：每台服务器8T磁盘，128G内存）

1). 每天日活跃用户100万，每人一天平均100条：100万*100条=1亿条；
2). 每条日志1k左右，每天1亿条：100000000 / 1024 /1024 = 约100G；
3). 半年内不扩容服务器：100G*180天=约18T；
4). 保存3个副本：18T*3=54T；
5). 预留20%～30%Buf = 54T/0.7 = 77T；
6). 约8T*10台服务器。

若是考虑数仓分层，数据压缩，又要怎么计算？

2.2.6 集群资源规划设计

在企业中通常会搭建一套生产集群和一套测试集群。生产集群运行生产任务，测试集群用于上线前代码编写和测试。

生产集群

1). 消耗内存的需分开；
2). 数据传输数据比较紧密的放在一起（Kafka、Zookeeper）；
3). 客户端尽量放在一到两台服务器上，方便外部访问；
4). 有依赖关系的尽量放在同一台服务器上（如Hive和Azkaban Executor）。

测试集群

三、数据生成模块

3.1 目标数据

我们要收集和分析数据主要包括页面数据、时间数据、曝光数据、启动数据和错误数据。

3.1.1 页面日志

页面数据主要记录一个页面的用户访问情况，包括访问时间、停留时间、页面路径等信息。

3.1.2 事件日志

时间数据主要记录应用内一个具体操作行为，包括操作类型
操作对象、操作对象描述等信息。

3.1.3 曝光日志

曝光数据主要记录页面所曝光的内容，包括曝光对象，曝光类型等信息。

3.1.4 启动日志

启动数据记录应用的启动信息。

3.1.5 错误日志

错误数据记录应用使用过程中的错误信息，包括错误编号及错误信息。

3.2数据埋点

3.2.1 主流埋点方式

目前主流的埋点方式，有代码埋点(前端/后端)、可视化埋点、全埋点三种。

代码埋点是通过调用埋点 SDK 函数，在需要埋点的业务逻辑功能位置调用接口，上报埋点数据。例如，我们对页面中的某个按钮埋点后，当这个按钮被点击时，可以在这个按钮对应的 OnClick 函数里面调用 SDK 提供的数据发送接口，来发送数据。

可视化埋点只需要研发人员集成采集 SDK，不需要写埋点代码，业务人员就可以通过访问分析平台的“圈选”功能，来“圈”出需要对用户行为进行捕捉的控件，并对该事件进行命名。圈选完毕后，这些配置会同步到各个用户的终端上，由采集 SDK 按照圈选的配置自动进行用户行为数据的采集和发送。

全埋点是通过在产品中嵌入 SDK，前端自动采集页面上的全部用户行为事件，上报埋点数据，相当于做了一个统一的埋点。然后再通过界面配置哪些数据需要在系统里面进行分析。

3.2.2 埋点数据上报时机

埋点数据上报时机包括两种方式：

方式一，在离开该页面时，上传在这个页面产生的所有数据(页面、事件、曝光、错误等)。
优点：批处理，减少了服务器接收数据压力，网络IO少；
缺点：实效性差。

方式二，每个事件、动作、错误等产生后，立即发送。
优点：响应及时；
缺点：对服务器接收数据压力比较大，网络IO增加。

本项目采用方式一埋点。

3.2.3 埋点数据日志结构

我们的日志结构大致可分为两类，一是普通页面埋点日志，二是启动日志。

普通页面日志结构如下，每条日志包含了，当前页面的页面信息，所有事件(动作)、所有曝光信息以及错误信息。除此之外，还包含了一系列公共信息，包括设备信息，地理位置，应用信息等，即下边的 common 字段。

3.3 服务器和JDK准备

分别安装 hadoop102、hadoop103、hadoop104 三台主机。
服务器和JDK的准备内容看这里==>

3.4 模拟数据

3.4.1 使用说明

创建 applog 路径

[xiaobai@hadoop102 module]$ mkdir applog

1). 将 application.yml、gmall2020-mock-log-2021-01-22.jar、path.json、logback.xml 上传到 hadoop102 的/opt/module/applog 目录下⬇️

2). 上传文件

配置文件⬇️

1). 配置application.yml文件：

# 外部配置打开
logging.config: "./logback.xml"
#业务日期
mock.date: "2020-06-14"#模拟数据发送模式
#mock.type: "http"
#mock.type: "kafka"
mock.type: "log"#http模式下，发送的地址
mock.url: "http://localhost:8080/applog"#kafka模式下，发送的地址
mock:kafka-server: "hadoop102:9092,hadoop103:9092,hadoop104:9092"kafka-topic: "ODS_BASE_LOG"#启动次数
mock.startup.count: 200
#设备最大值
mock.max.mid: 500000
#会员最大值
mock.max.uid: 100
#商品最大值
mock.max.sku-id: 35
#页面平均访问时间
mock.page.during-time-ms: 20000
#错误概率 百分比
mock.error.rate: 3
#每条日志发送延迟 ms
mock.log.sleep: 10
#商品详情来源  用户查询，商品推广，智能推荐, 促销活动
mock.detail.source-type-rate: "40:25:15:20"
#领取购物券概率
mock.if_get_coupon_rate: 75
#购物券最大id
mock.max.coupon-id: 3
#搜索关键词
mock.search.keyword: "图书,小米,iphone11,电视,口红,ps5,苹果手机,小米盒子"

2). 配置path.json文件：

[{"path":["home","good_list","good_detail","cart","trade","payment"],"rate":20 },{"path":["home","search","good_list","good_detail","login","good_detail","cart","trade","payment"],"rate":40 },{"path":["home","mine","orders_unpaid","trade","payment"],"rate":10 },{"path":["home","mine","orders_unpaid","good_detail","good_spec","comment","trade","payment"],"rate":5 },{"path":["home","mine","orders_unpaid","good_detail","good_spec","comment","home"],"rate":5 },{"path":["home","good_detail"],"rate":10 },{"path":["home"  ],"rate":10 }

3). 配置logback文件：

<?xml version="1.0" encoding="UTF-8"?>
<configuration><property name="LOG_HOME" value="/opt/module/applog/log" /><appender name="console" class="ch.qos.logback.core.ConsoleAppender"><encoder><pattern>%msg%n</pattern></encoder></appender><appender name="rollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender"><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><fileNamePattern>${LOG_HOME}/app.%d{yyyy-MM-dd}.log</fileNamePattern></rollingPolicy><encoder><pattern>%msg%n</pattern></encoder></appender><!-- 将某一个包下日志单独打印日志 --><logger name="com.xiaobai.gmall2020.mock.log.util.LogUtil"level="INFO" additivity="false"><appender-ref ref="rollingFile" /><appender-ref ref="console" /></logger><root level="error"  ><appender-ref ref="console" /></root>
</configuration>
~

生成日志

1). 如图，在/opt/module/applog目录下执行以下命令生成对应的日志文件log：

 java -jar gmall2020-mock-log-2021-01-22.jar

2). 在/opt/module/applog/log目录下查看生成日志：

3.4.2 集群日志生成脚本

在 /home/xiaobai/bin目录下创建脚本lg.sh：

[xiaobai@hadoop102 bin]$ vim lg.sh
#!/bin/bashfor i in hadoop102 hadoop103; do
echo "========== $i =========="
ssh $i "cd /opt/module/applog/; java -jar gmall2020-mock-log-2021-01-22.jar >/dev/null 2>&1 &"
done

注:
1). /opt/module/applog/为 jar 包及配置文件所在路径
2). /dev/null 代表 linux 的空设备文件，所有往这个文件里面写入的内容都会丢失，俗
称“黑洞”。
标准输入 0:从键盘获得输入 /proc/self/fd/0;
标准输出 1:输出到屏幕(即控制台) /proc/self/fd/1;
错误输出 2:输出到屏幕(即控制台) /proc/self/fd/2。

修改脚本执行权限:

[xiaobai@hadoop102 bin]$ chmod 777 lg.sh

将applog分发给其他节点：

[xiaobai@hadoop102 module]$ xsync applog/

删除hadoop104上的applog以及hadoop102上的log，并测试lg.sh：

[xiaobai@hadoop104 module]$ rm -rf applog/
[xiaobai@hadoop102 applog]$ rm -rf log

如图，hadoop102 / hadoop103加载出了log：

四、数据采集模块

4.1 集群所有进程查看脚本

在/home/xiaobai/bin 目录下创建脚本 xcall.sh:

[xiaobai@hadoop102 bin]$ vim xcall.sh
#! /bin/bashfor i in hadoop102 hadoop103 hadoop104
do
echo --------- $i ----------ssh $i "$*"
done

修改脚本执行权限：

[xiaobai@hadoop102 bin]$ chmod 777 xcall.sh

启动脚本：

[xiaobai@hadoop102 bin]$ xcall.sh jps
--------- hadoop102 ----------
3081 Jps
--------- hadoop103 ----------
2961 Jps
--------- hadoop104 ----------
3295 Jps

4.2 zookeeper安装

4.2.1 安装ZK

集群规划
在hadoop102、hadoop103、hadoop104三个节点上部署zookeeper。
解压安装

1). 解压zookeeper安装包到/opt/module/目录下：

[xiaobai@hadoop102 software]$ tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/

2). 修改/opt/module/apache-zookeeper-3.5.7-bin名称为zookeeper-3.5.7：

[xiaobai@hadoop102 module]$ mv apache-zookeeper-3.5.7-bin/ zookeeper-3.5.7

配置服务器编号：

1). 在/opt/module/zookeeper-3.5.7目录下创建zkData：

[xiaobai@hadoop102 zookeeper-3.5.7]$ mkdir zkData

2). 在/opt/module/zookeeper-3.5.7/zkData目录下创建一个myid文件：

[xiaobai@hadoop102 zookeeper-3.5.7]$ vim myid

添加myid文件，⚠️一定要在Linux里面创建！
在文件中添加域server对应的编号：

3). 同步/opt/module/zookeeper-3.5.7目录到hadoop103、hadop104:

[xiaobai@hadoop102 module]$ xsync zookeeper-3.5.7/

分别在hadoop103、hadoop104上修改myid文件中内容为3、4:

[xiaobai@hadoop103 zkData]$ vim myid
3
[xiaobai@hadoop104 zkData]$ vim myid
4

配置zoo.cfg文件：

1). 重命名/opt/module/zookeeper-3.5.7/conf目录下的zoo_sample.cfg为zoo.cfg ：

[xiaobai@hadoop102 conf]$ mv zoo_sample.cfg zoo.cfg

2). 打开zoo.cfg 文件：

[xiaobai@hadoop102 conf]$ vim zoo.cfg

如图，修改数据存储路径配置：

dataDir=/opt/module/zookeeper-3.5.7/zkData

如图，增加如下配置：

#######################cluster###################################
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

3). 同步分发zoo.cfg 配置文件：

[xiaobai@hadoop102 conf]$ xsync zoo.cfg

4). 配置参数解读：

server.A=B:C:D

A 是一个数字，表示这是第几号服务器;

集群模式下配置一个文件 myid，这个文件在 dataDir 目录下，这个文件里面有一个数据就是 A 的值，Zookeeper 启动时读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比较从而判断到底是哪个 server;

B 是这个服务器的地址;

C 是这个服务器 Follower 与集群中的 Leader 服务器交换信息的端口;

D 是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

集群操作

1). 分别启动zookeeper：

[xiaobai@hadoop102 zookeeper-3.5.7]$ bin/zkServer.sh start
[xiaobai@hadoop103 zookeeper-3.5.7]$ bin/zkServer.sh start
[xiaobai@hadoop104 zookeeper-3.5.7]$ bin/zkServer.sh start

2). 查看状态：

[xiaobai@hadoop102 zookeeper-3.5.7]$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower[xiaobai@hadoop103 zookeeper-3.5.7]$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower[xiaobai@hadoop104 zookeeper-3.5.7]$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader

4.2.2 ZK集群启动停止脚本

在hadoop102的/home/xiaobai/bin目录下创建脚本zk.sh：

在脚本中编写如下内容:

#!/bin/bashcase $1 in
"start")for i in hadoop102 hadoop103 hadoop104doecho "---------- $i ------------"ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"done
;;
"stop")for i in hadoop102 hadoop103 hadoop104doecho "---------- $i ------------"ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"done
;;
"status")for i in hadoop102 hadoop103 hadoop104doecho "---------- $i ------------"ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"done
;;
esac

增加脚本执行权限：

[xiaobai@hadoop102 bin]$ chmod 777 zk.sh

Zookeeper 集群启动脚本：

[xiaobai@hadoop102 ~]$ zk.sh start

Zookeeper 集群停止脚本：

[xiaobai@hadoop102 ~]$ zk.sh stop

查看Zookeeper 状态：

[xiaobai@hadoop102 ~]$ zk.sh status
---------- hadoop102 ------------
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader
---------- hadoop103 ------------
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower
---------- hadoop104 ------------
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower

4.3 Kafka安装

4.3.1 Kafka集群安装

解压缩kafka_2.11-2.4.1.tgz至/opt/module/目录下：

[xiaobai@hadoop102 software]$ tar -zxvf kafka_2.11-2.4.1.tgz -C /opt/module/

修改kafka_2.11-2.4.1名称为kafka：

[xiaobai@hadoop102 module]$ mv kafka_2.11-2.4.1/ kafka

在/opt/module/kafka路径下创建logs文件夹：

[xiaobai@hadoop102 kafka]$ mkdir logs

如图，在/opt/module/kafka/config目录下修改配置文件server.properties：

[xiaobai@hadoop102 config]$ vim server.properties #broker 的全局唯一编号，不能重复 broker.id=0
#删除 topic 功能使能
delete.topic.enable=true
#kafka 运行日志存放的路径
log.dirs=/opt/module/kafka/logs
#配置连接 Zookeeper 集群地址 zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181

分发kafka：

[xiaobai@hadoop102 module]$ xsync kafka/

修改各自的broker.id：

[xiaobai@hadoop103 config]$ vim server.properties
broker.id=1
[xiaobai@hadoop104 config]$ vim server.properties
broker.id=2

配置环境变量：

[xiaobai@hadoop102 kafka]$ sudo vim /etc/profile.d/my_env.sh
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:KAFKA_HOME/bin[xiaobai@hadoop102 kafka]$ source /etc/profile.d/my_env.sh

分发my_env.sh：

[xiaobai@hadoop102 kafka]$ sudo /home/xiaobai/bin/xsync /etc/profile.d/my_env.sh

在hadoop103、hadoop104上使用source刷新my_env.sh ：

[xiaobai@hadoop103 config]$ source /etc/profile.d/my_env.sh
[xiaobai@hadoop104 config]$ source /etc/profile.d/my_env.sh

启动集群：
依次在 hadoop102、hadoop103、hadoop104 节点上启动 kafka：

[xiaobai@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties
[xiaobai@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties
[xiaobai@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties

关闭集群：
在/opt/module/kafka目录下执行以下命令：

bin/kafka-server-stop.sh stop

4.3.2 Kafka集群启动停止脚本

在/home/xiaobai/bin目录下创建脚本文件kf.sh：

#!/bin/bashcase $1 in
"start")for i in hadoop102 hadoop103 hadoop104doecho "------启动 $i KAFKA------"ssh $i "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"done
;;
"stop")for i in hadoop102 hadoop103 hadoop104doecho "------停止 $i KAFKA------"                ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh stop"done
;;
esac

赋予权限：

[xiaobai@hadoop102 bin]$ chmod 777 kf.sh

使用脚本开启kafka：

[xiaobai@hadoop102 bin]$ kf.sh start

使用脚本停止kafka：

[xiaobai@hadoop102 bin]$ kf.sh stop

4.3.3 常用命令

查看Kafka Topic列表:

[xiaobai@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka --list
__consumer_offsets
topic_log

创建Kafka Topic:
/opt/module/kafka/目录下创建日志主题:

[xiaobai@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka  --create --replication-factor 1 --partitions 1 --topic topic_log

删除Kafka Topic:

[xiaobai@hadoop102 kafka]$ bin/kafka-topics.sh --delete --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka --topic topic_log

Kafka生产消息:

[xiaobai@hadoop102 kafka]$ bin/kafka-console-producer.sh \
> --broker-list hadoop102:9092 --topic topic_log
>hello world
>xiaobai is coming
>work

Kafka消费消息:

[xiaobai@hadoop102 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop102:9092 --from-beginning --topic topic_log
hello world
xiaobai is coming
work

注⚠️：–from-beginning：会把主题中以往所有的数据都读取出来。根据业务场景选择是否增加该配置。

查看Kafka Topic详情:

[xiaobai@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka --describe --topic topic_log

4.3.4 项目经验之Kafka机器数量计算

Kafka机器数量（经验公式）= 2 *（峰值生产速度 * 副本数 / 100）+ 1；

先拿到峰值生产速度，再根据设定的副本数，就能预估出需要部署Kafka的数量。

峰值生产速度：可以通过压测得到；
副本数：

默认副本数是1个，在企业里会有2-3个，通常为2；
副本多可以提高可靠性，但是会降低网络传输速率；
如峰值生产速度为50M/s，副本数为2；
Kafka机器数量 = 2 * （50 * 2 / 100）+ 1 = 3台。

4.3.5 项目经验之压力测试

Kafka压测:

用Kafka官方自带的脚本，对Kafka进行压测：

kafka-consumer-perf-test.sh
kafka-producer-perf-test.sh

Kafka压测时，在硬盘读写速度一定的情况下，可以查看到哪些地方出现了瓶颈(CPU，内存，网络 IO)；一般都是网络 IO 达到瓶颈。

4.3.6 项目经验之Kafka分区数计算

创建一个只有 1 个分区的 topic;
测试这个 topic 的 producer 吞吐量和 consumer 吞吐量;
假设他们的值分别是 Tp 和 Tc，单位可以是 MB/s;
然后假设总的目标吞吐量是 Tt，那么分区数=Tt / min(Tp，Tc) 例如:producer 吞吐量=20m/s;
consumer 吞吐量=50m/s，期望吞吐量 100m/s;
分区数=100 / 20 =5 分区;
分区数一般设置为:3-10 个。
建议参考==》

4.4采集日志Flume

4.4.1 日至采集Flume安装

解压 apache-flume-1.9.0-bin.tar.gz 到/opt/module/目录下:

[xiaobai@hadoop102 software]$ tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /opt/module/

修改apache-flume-1.9.0-bin的名称为 flume

[xiaobai@hadoop102 module]$ mv apache-flume-1.9.0-bin/ flume

将lib文件夹下的guava-11.0.2.jar 删除以兼容hadoop3.2.2：

[xiaobai@hadoop102 lib]$ rm -rf guava-11.0.2.jar

注⚠️：删除guava-11.0.2.jar 的服务器节点，一定要配置hadoop环境变量，否则会报异常。

将 flume/conf 下的 flume-env.sh.template 文件修改为 flume-env.sh，并配置flume-env.sh 文件：

[xiaobai@hadoop102 conf]$ mv flume-env.sh.template flume-env.sh
[xiaobai@hadoop102 conf]$ vim flume-env.sh
export JAVA_HOME=/opt/module/jdk1.8.0_144

分发flume：

[xiaobai@hadoop102 module]$ xsync flume/

4.4.2 项目经验之Flume组件选型

source

1). flume1.6 source:

exec :
优点：可以实时监控文件变化；
缺点：有丢数据的风险。

spooling：
优点：可以实现断点续传；
缺点：不能实时监控文件变化。

2). flume 1.7 taildir source⭐️

Taildir Source 相比 Exec Source、Spooling Directory Source 的优势:
a. 可实现断点续传,多目录；Flume1.6 以前需要自己自定义 Source 记录每次读取文件位置，实现断点续传。
b. 可以实时监控文件的变化。
Exec Source 可以实时搜集数据，但是在 Flume 不运行或者 Shell 命令出错的情况下，数据将会丢失；Spooling Directory Source 监控目录，支持断点续传，但不能实时监控文件变化。

3). batchSize 大小如何设置?

答:Event 1K 左右时，500-1000 合适(默认为 100)

Channel

file channel⭐️： 数据存储在磁盘中，可靠性高，效率低；
memory channel⭐️： 数据存储在内存中，效率高，可靠性差。

采用 Kafka Channel，省去了 Sink，提高了效率。KafkaChannel 数据存储在 Kafka 里面，所以数据是存储在磁盘中。

4.4.3 日志采集 Flume 配置

Flume配置分析：

注⚠️：Flume 直接读 log 日志的数据，log 日志的格式是 app.yyyy-mm-dd.log。
Flume的具体配置如下：

1). 在/opt/module/flume/conf 目录下创建 file-flume-kafka.conf 文件:

[xiaobai@hadoop102 conf]$ vim file-flume-kafka.conf

2). 在文件配置如下内容:

#定义组件
a1.sources=r1
a1.channels=c1#配置source (Taildirsource)
a1.sources.r1.type=TAILDIR
a1.sources.r1.filegroups=f1
a1.sources.r1.filegroups.f1=/opt/module/applog/log/app.*
a1.sources.r1.positionFile=/opt/module/flume/taildir_position.json#配置拦截器（ETL数据清洗 判断json是否完整）
a1.sources.r1.interceptors=i1
a1.sources.r1.interceptors.i1.type=com.xiaobai.flume.interceptor.ETLInterceptor$Builder#配置channel
a1.channels.c1.type=org.apache.flume.channel.kafka.kafkaChannel
a1.channels.c1.kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092
a1.channels.c1.kafka.topic=topic_log
a1.channels.c1.parseAsFlumeEvent=false#配置sink(Kafka Channel 无Sink)#拼接组件
a1.sources.r1.channels=c1

3). 分发file-flume-kafka.conf：

[xiaobai@hadoop102 conf]$ xsync file-flume-kafka.conf

tips: com.xiaobai.flume.interceptor.ETLInterceptor是自定义的拦截器全类名。

4.4.4 Flume拦截器

创建 Maven 工程 flume-interceptor;
创建包名: com.xiaobai.flume.interceptor;
在 pom.xml 文件中添加如下配置:

<properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version><scope>provided</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency></dependencies><build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>

在com.xiaobai.flume.interceptor包下创建JSONUtils类：

package com.xiaobai.flume.interceptor;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;public class JSONUtils {//    //test数据
//    public static void main(String[] args) {//        System.out.println(isValidate("{123323"));
//    }//验证数据是否jsonpublic static boolean isValidate(String log) {try {JSON.parse(log);return true;} catch(JSONException e){return false;}}
}

在com.xiaobai.flume.interceptor下创建ETLInterceptor类：

package com.xiaobai.flume.interceptor;import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.List;public class ETLInterceptor implements Interceptor {@Overridepublic void initialize() {}@Overridepublic Event intercept(Event event) {// 取数据后进行校验//1 获取数据byte[] body = event.getBody();String log = new String(body, Charset.forName("UTF-8"));//2 校验if(JSONUtils.isValidate(log)){return event;}return null;}@Overridepublic List<Event> intercept(List<Event> list) {Iterator<Event> iterator = list.iterator();while(iterator.hasNext()){Event next = iterator.next();if(intercept(next) == null){iterator.remove();}}return list;}public static class Builder implements Interceptor.Builder{@Overridepublic Interceptor build() {return new ETLInterceptor();}@Overridepublic void configure(Context context) {}}@Overridepublic void close() {}
}

打包：
将打包好的带依赖的jar包传入hadoop102 的/opt/module/flume/lib 目录下：
过滤查找是否存在带依赖的jar包flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar：

[xiaobai@hadoop102 lib]$ ls | grep interceptor
flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

分发jar包至hadoop103、hadoop104：

[xiaobai@hadoop102 lib]$ xsync flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

分别在 hadoop102、hadoop103 上启动 Flume：

[xiaobai@hadoop102 flume]$ bin/flume-ng agent --name a1 --conf-file conf/file-flume-kafka.conf &
[xiaobai@hadoop103 flume]$ bin/flume-ng agent --name a1 --conf-file conf/file-flume-kafka.conf &

4.4.5 测试Flume-Kafka通道

生成日志：

[xiaobai@hadoop102 kafka]$ lg.sh

消费Kafka数据，观察控制台是否获取到数据：

[xiaobai@hadoop102 kafka]$ bin/kafka-console-consumer.sh \--bootstrap-server hadoop102:9092 --from-beginning --topic topic_log

如图所示就是已经获取到数据了！我这里用两个远程工具同时开了好几个远程端口进行不同操作，所以这个端口看起来会不一样。

注⚠️：
这里获取不到数据的话，后面会很麻烦！！！！
前车之鉴，一定要认真检查！配置文件！少个字母多个字母包名不正确之类的都会引起错误！！！若是配置文件都没错的话就检查Kafka、Flume、Zookeeper是否正确启动，仔细检查状态！！！再检查Flume的拦截器代码是否正常！！所引用的包名是否正确！！！！！！！！试图引起注意！！！！！！！！n多个感叹号❕足以证明会有多重要我是当时不以为然没在意结果到后面进行不下去了又返回来重新检查了一遍的

数据仓库之电商数仓-- 1、用户行为数据采集相关推荐

数据仓库之电商数仓-- 2、业务数据采集平台
目录一.电商业务简介 1.1 电商业务流程 1.2 电商常识(SKU.SPU) 1.3 电商系统表结构 1.3.1 活动信息表(activity_info) 1.3.2 活动规则表(activity ...
31 大数据项目之电商数仓（用户行为数据采集）
文章目录第1章大数据项目之电商数仓(用户行为数据采集) 第2章项目需求 2.1 项目需求分析 2.2 项目框架 2.2.1 技术选型 2.2.2 系统架构图设计 2.2.3 系统数据流程设计 2 ...
大数据项目之电商数仓（用户行为数据采集）
第1章数据仓库概念第2章项目需求 2.1 项目需求分析 2.2 项目框架 2.2.1 技术选型 2.2.2 系统架构图设计 2.2.3 系统数据流程设计 2.2.4 框架版本选型产品版本 H ...
数据仓库之电商数仓-- 4、可视化报表Superset
目录一.Superset入门 1.1 Superset概述 1.2 Superset应用场景二.Superset安装及使用 2.1 安装Python环境 2.1.1 安装Miniconda 2.1 ...
数据仓库之电商数仓-- 3.4、电商数据仓库系统(ADS层)
目录九.数仓搭建-ADS层 9.1 建表说明 9.2 访客主题 9.2.1 访客统计 9.2.2 路径分析 9.3 用户主题 9.3.1 用户统计 9.3.2 用户变动统计 9.3.3 用户行为漏斗 ...
数据仓库之电商数仓-- 3.3、电商数据仓库系统(DWT层)
目录八.数仓搭建-DWT层 8.1 访客主题 8.2 用户主题 8.3 商品主题 8.4 优惠券主题 8.5 活动主题 8.6 地区主题 8.7 DWT层首日数据导入脚本 8.8 DWT层每日数据导 ...
数据仓库之电商数仓-- 3.2、电商数据仓库系统(DWS层)
目录七.数仓搭建-DWS层 7.1 系统函数 7.1.1 nvl函数 7.1.2 日期处理函数 7.1.3 复杂数据类型定义 7.2 DWS层 7.2.1 访客主题 7.2.2 用户主题 7.2.3 ...
数据仓库之电商数仓-- 5、即席查询Kylin
目录一.Kylin 1.1 Kylin简介 1.1.1 Kylin定义 1.1.2 Kylin相关术语 1.1.3Kylin架构 1.1.4 Kylin特点 1.2 Kylin安装 1.2.1 Ky ...
1、电商数仓（用户行为采集平台）
数据仓库概念数据仓库( Data Warehouse ),是为企业制定决策,提供数据支持的.可以帮助企业,改进业务流程.提高产品质量等. 数据仓库的输入数据通常包括:业务数据.用户行为数据和爬虫数据 ...

数据仓库之电商数仓-- 1、用户行为数据采集

目录