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

目录

• 一、数据仓库概念
• 二、项目需求及架构设计
• • 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 项目经验之Kafka机器数量计算
    • 4.3.4 项目经验之压力测试
    • 4.3.5 项目经验之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页面不能显示完整信息

一、数据仓库概念

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

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

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

二、项目需求及架构设计 2.1 项目需求分析

项目需求

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

2.2 项目框架 2.2.1 技术选型

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

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

数据采集传输：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 框架版本选型

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

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

2. 云服务选择

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

3. 具体版本型号

Apache框架版本：

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

2.2.4 服务器选型

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

2.2.5 集群规模

1. 如何确认集群规模？（假设：每台服务器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.2.6 集群资源规划设计

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

1. 生产集群

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

2. 测试集群

三、数据生成模块 3.1 目标数据

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

3.1.1 页面日志

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

3.1.2 事件日志

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

3.1.3 曝光日志

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

Y2Vf,size_20,color_FFFFFF,t_70,g_se,x_16)

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 使用说明

1. 将 application.properties、gmall2020-mock-log-2020-05-10.jar、path.json、logback.xml 上传到 hadoop102 的/opt/module/applog 目录下

1). 创建 applog 路径

[xiaobai@hadoop102 module]$ mkdir applog

2). 上传文件

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

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

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

3.4.2 集群日志生成脚本

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

[xiaobai@hadoop102 bin]$ vim lg.sh
#!/bin/bash

for 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。

2. 修改脚本执行权限:

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

3. 将applog分发给其他节点：

[xiaobai@hadoop102 module]$ xsync applog/

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

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

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

四、数据采集模块 4.1 集群所有进程查看脚本

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

[xiaobai@hadoop102 bin]$ vim xcall.sh
#! /bin/bash

for i in hadoop102 hadoop103 hadoop104 
do
echo --------- $i ----------
   ssh $i "$*"
done

2. 修改脚本执行权限：

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

3. 启动脚本：

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

4.2 zookeeper安装 4.2.1 安装ZK

1. 集群规划
   在hadoop102、hadoop103、hadoop104三个节点上部署zookeeper。

2. 解压安装

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

3. 配置服务器编号：

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对应的编号：

2

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

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，而这个端口就是用来执行选举时服务器相互通信的端口。

5. 集群操作

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集群启动停止脚本

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

在脚本中编写如下内容:

#!/bin/bash

case $1 in
"start")
        for i in hadoop102 hadoop103 hadoop104
        do
                echo "---------- $i ------------"
                ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
        done
;;
"stop")
        for i in hadoop102 hadoop103 hadoop104
        do
                echo "---------- $i ------------"
                ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
        done
;;
"status")
        for i in hadoop102 hadoop103 hadoop104
        do
                echo "---------- $i ------------"
                ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
        done
;;
esac

2. 增加脚本执行权限：

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

3. Zookeeper 集群启动脚本：

[xiaobai@hadoop102 ~]$ zk.sh start

4. Zookeeper 集群停止脚本：

[xiaobai@hadoop102 ~]$ zk.sh stop

5. 查看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集群安装

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

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

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

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

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

[xiaobai@hadoop102 kafka]$ mkdir logs

4. 如图，在/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

5. 分发kafka：

[xiaobai@hadoop102 module]$ xsync kafka/

6. 修改各自的broker.id：

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

7. 配置环境变量：

[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 

8. 分发my_env.sh：

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

9. 在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 

10. 启动集群：
    依次在 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 

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

bin/kafka-server-stop.sh stop

4.3.2 Kafka集群启动停止脚本

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

#!/bin/bash

case $1 in
"start")
        for i in hadoop102 hadoop103 hadoop104
        do
                echo "------启动 $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 hadoop104
        do
                echo "------停止 $i KAFKA------"                ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh"
        done
;;
esac

2. 赋予权限：

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

3. 使用脚本开启kafka：

[xiaobai@hadoop102 bin]$ kf.sh start

4. 使用脚本停止kafka：

[xiaobai@hadoop102 bin]$ kf.sh stop

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

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

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

1. 峰值生产速度：可以通过压测得到；

2. 副本数：

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

4.3.4 项目经验之压力测试

Kafka压测:

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

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

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

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

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

4.4采集日志Flume 4.4.1 日至采集Flume安装

1. 解压 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/

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

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

3. 将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环境变量，否则会报异常。

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

5. 分发flume：

[xiaobai@hadoop102 module]$ xsync flume/

4.4.2 项目经验之Flume组件选型

1. 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)

2. Channel

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

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

4.4.3 日志采集 Flume 配置

1. Flume配置分析：
   
   注：Flume 直接读 log 日志的数据，log 日志的格式是 app.yyyy-mm-dd.log。

2. 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.source.r1.type=TAILDIR
a1.source.r1.filegroups=f1
a1.source.r1.filegroups.f1=/opt/module/applog/log/app.*
a1.source.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拦截器

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

        8
        8
    


    
        
            org.apache.flume
            flume-ng-core
            1.9.0
            provided
        

        
            com.alibaba
            fastjson
            1.2.62
        
    

    
        
            
                maven-compiler-plugin
                2.3.2
                
                    1.8
                    1.8
                
            
            
                maven-assembly-plugin
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
        
    

4. 在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"));
//    }

    //验证数据是否json
    public static boolean isValidate(String log) {

        try {

            JSON.parse(log);
            return true;
        } catch(JSONException e){
            return false;
        }
    }
}

5. 在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 {
    @Override
    public void initialize() {

    }

    @Override
    public 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;
    }

    @Override
    public List intercept(List list) {

        Iterator 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{
        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
     @Override
    public void close() {


    }
}

6. 打包：
   

7. 将打包好的带依赖的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

8. 分发jar包至hadoop103、hadoop104：

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

9. 分别在 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通道

1. 生成日志：

[xiaobai@hadoop102 kafka]$ lg.sh

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

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

获取不到数据，检查Kafka、Flume、Zookeeper是否正确启动，检查Flume的拦截器代码是否正常。

4.4.6 日至采集Flume启动停止脚本

1. 创建日至采集Flume启动停止脚本f1.sh：

[xiaobai@hadoop102 bin]$ vim f1.sh

2. 填入以下内容：

#!/bin/bash

case $1 in
"start")
        for i in hadoop102 hadoop103
        do
                echo "------启动$i采集flume------"
                ssh $i "nohup /opt/module/flume/bin/flume-ng agent --conf-file /opt/module/flume/conf/file-flume-kafka.conf --name a1 -Dflume.root.logger=INFO,LOGFILE >/opt/module/flume/log1.txt 2>&1 &"
        done
;;
"stop")
      for i in hadoop102 hadoop103
        do
                echo "------停止$i采集flume------"                ssh $i "ssh $i ps -ef | grep file-flume-kafka.conf | grep -v grep | awk '{print $2}' | xargs -n1 kill -9"
        done
;;
esac

3. 增加脚本执行权限：

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

4. f1集群启动脚本：

[xiaobai@hadoop102 bin]$ f1.sh start

5. f1集群停止脚本：

[xiaobai@hadoop102 bin]$ f1.sh stop

⭐️注：

1. nohup，该命令可以在你退出帐户/关闭终端之后继续运行相应的进程。nohup 就是不挂起的意思，不挂断地运行命令;
2. awk 默认分隔符为空格;
3. xargs 表示取出前面命令运行的结果，作为后面命令的输入参数;
4. $2是在双引号 “ ”呢句会被解析为脚本的第二个参数，但此处表达的意思是awk的第二个值，所以需要转义，用$2表示。

4.5 消费Kafka数据Flume 4.5.1 项目经验之Flume组件选型 4.5.1.1 FileChannel 和 MemoryChannel 区别

MemoryChannel 传输数据速度更快，但因为数据保存在 JVM 的堆内存中，Agent 进程挂掉会导致数据丢失，适用于对数据质量要求不高的需求。

FileChannel 传输速度相对于 Memory 慢，但数据安全保障高，Agent 进程挂掉也可以从失败中恢复数据。

选型:
金融类公司、对钱要求非常准确的公司通常会选择 FileChannel；
传输的是普通日志信息(京东内部一天丢 100 万-200 万条，这是非常正常的)，通常选择 MemoryChannel。

4.5.1.2 FileChannel优化

通过配置 dataDirs 指向多个路径，每个路径对应不同的硬盘，增大 Flume 吞吐量。 官方说明如下:
Comma separated list of directories for storing log files. Using multiple directories on separate disks can improve file channel peformance

checkpointDir 和 backupCheckpointDir 也尽量配置在不同硬盘对应的目录中，保证checkpoint 坏掉后，可以快速使用 backupCheckpointDir 恢复数据。

FileChannel底层原理

4.5.1.3 Sink：HDFS Sink

1. HDFS 存入大量小文件，有什么影响?

元数据层面: 每个小文件都有一份元数据，其中包括文件路径，文件名，所有者，所属组，权限，创建时间等，这些信息都保存在 Namenode 内存中。所以小文件过多，会占用 Namenode 服务器大量内存，影响 Namenode 性能和使用寿命；

计算层面:默认情况下 MR 会对每个小文件启用一个 Map 任务计算，非常影响计算性能。同时也影响磁盘寻址时间。

2. HDFS 小文件处理

官方默认的这三个参数配置写入 HDFS 后会产生小文件，hdfs.rollInterval、hdfs.rollSize、 hdfs.rollCount

基于以上 hdfs.rollInterval=3600，hdfs.rollSize=134217728，hdfs.rollCount =0 几个参数综合作用，效果如下:
1). 文件在达到 128M 时会滚动生成新文件;
2). 文件创建超 3600 秒时会滚动生成新文件.

4.5.2 Flume时间戳拦截器

由于 flume 默认会用 linux 系统时间，作为输出到 HDFS 路径的时间。如果数据是 23:59 分产生的。Flume 消费 kafka 里面的数据时，有可能已经是第二天00:00以后了，那么该部门数据会被发往第二天的 HDFS 路径。我们希望的是根据日志里面的实际时间，发往 HDFS 的路径，所以下面拦截器作用是获取日志中的实际时间。

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

package com.xiaobai.flume.interceptor;

import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

public class TimeStampInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        //将日志拦截，取出header里面的key，取出body里面对应的时间，将ts的值赋值给header里的key timestamp

        //1. 获取header头
        Map headers = event.getHeaders();

        //2. 获取body中的ts
        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);

        JSONObject jsonObject = JSONObject.parseObject(log);
        String ts = jsonObject.getString("ts");
        
        //3. 将ts赋值给timestamp
        headers.put("timestamp",ts);
        
        return event;
    }

    @Override
    public List intercept(List list) {

        for (Event event : list) {
            intercept(event);
            
        }
        
        return list;
    }

    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new TimeStampInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

    @Override
    public void close() {

    }
}

2. 重新打包：

3. 需要先将打好的包放入到 hadoop104 的/opt/module/flume/lib 文件夹下面:

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

4.5.3 消费者Flume配置

1. Flume配置分析：
   
2. 具体配置如下：

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

[xiaobai@hadoop104 conf]$ vim kafka-flume-hdfs.conf

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

## 定义组件
## 定义组件 
a1.sources=r1
a1.channels=c1
a1.sinks=k1

## 配置source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
#a1.sources.r1.batchSize = 5000 a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sources.r1.kafka.topics=topic_log
a1.sources.r1.batchSize=5000
a1.sources.r1.batchDurationMills=2000


#时间戳拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.xiaobai.flume.interceptor.TimeStampInterceptor$Builder

## 配置channel
a1.channels.c1.type = file
a1.channels.c1.checkpointDir=/opt/module/flume/checkpoint/behavior1
a1.channels.c1.dataDirs=/opt/module/flume/data/behavior1/
#a1.channels.c1.maxFileSize = 2146435071
#a1.channels.c1.capacity = 1000000 a1.channels.c1.keep-alive = 6

## 配置sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path =/origin_data/gmall/log/topic_log/%Y-%m-%d

a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.hdfs.rollSize=134217728
a1.sinks.k1.hdfs.rollCount=0

a1.sinks.k1.hdfs.filePrefix = log-
a1.sinks.k1.hdfs.round = false
#a1.sinks.k1.hdfs.fileType=DataStream
##a1.sinks.k1..hdfs.writeFormat=Text
#
#
#a1.sinks.k1.hdfs.rollInterval = 10 
#a1.sinks.k1.hdfs.rollSize = 134217728 
#a1.sinks.k1.hdfs.rollCount = 0

## 控制输出文件是原生文件
a1.sinks.k1.hdfs.fileType = CompressedStream a1.sinks.k1.hdfs.codeC = lzop

## 拼接组件 
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1

4.5.4 消费者Flume启动停止脚本

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

[xiaobai@hadoop102 bin]$ vim f2.sh

在脚本中填写如下内容：

#!/bin/bash

case $1 in
"start"){
        for i in hadoop104
        do
                echo "------启动$i消费flume------"
                ssh $i "nohup /opt/module/flume/bin/flume-ng agent --conf-file /opt/module/flume/conf/kafka-flume-hdfs.conf --name a1 -Dflume.root.logger=INFO,LOGFILE >/opt/module/flume/log2.txt 2>&1 &"
        done
};;
"stop"){
      for i in hadoop104
        do
                echo "------停止$i消费flume------"
                ssh $i "ssh $i ps -ef | grep kafka-flume-hdfs | grep -v grep | awk '{print $2}' | xargs -n1 kill"
        done
};;
esac

3. 增加脚本执行权限：

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

4. f2 集群启动脚本：

[xiaobai@hadoop102 bin]$ f2.sh start

5. f2 集群停止脚本：

[xiaobai@hadoop102 bin]$ f2.sh stop

4.6 采集通道启动停止脚本

1. 在/home/atguigu/bin 目录下创建脚本 cluster.sh:

[xiaobai@hadoop102 bin]$ vim cluster.sh

2. 填入以下内容：

#!/bin/bash


case $1 in
"start")
        echo "------启动集群------"
        zk.sh start
        myhadoop.sh start
        kf.sh start
        f1.sh start
        f2.sh start

;;
"stop")
        echo "------停止集群------"

        f2.sh stop

        f1.sh stop

        kf.sh stop

        myhadoop.sh stop

        zk.sh stop
;;
esac

3. 增加脚本执行权限:

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

4. cluster 集群启动脚本:

[xiaobai@hadoop102 bin]$ cluster.sh start

5. cluster 集群停止脚本:

五、常见问题及解决方案 5.1 2NN页面不能显示完整信息

1. 问题描述：
如图，访问2NN页面http://hadoop104:9868,看不到详细信息。

2. 解决方法：

1. 在浏览器上按command+shift+i进入以下界面，点击console会有错误提示信息：
   
2. 如图，找到要修改的文件：

[xiaobai@hadoop104 static]$ pwd
/opt/module/hadoop-3.2.2/share/hadoop/hdfs/webapps/static

[xiaobai@hadoop104 static]$ vim dfs-dust.js

输入以下代码显示行号：

:set nu

如图，shift+d:修改替换61行：

return new Date(Number(v)).toLocalString();

3. 分发dfs-dust.js:

[xiaobai@hadoop104 static]$ xsync dfs-dust.js

4. 由于此界面是静态网页，所以需强制刷新http://hadoop104:9868页面，点击浏览器三个小点–>更多工具–>清除浏览数据–清除数据，显示以下界面：
   

3. 问题描述：
hadoop104拒绝连接请求。

4. 解决方法：
如果网络畅通且关闭了防火墙，那可能是由于集群忘记开启，重新启动集群再次刷新hadoop104:9868界面即可。