您的当前位置：首页数据仓库概念完整使用 (第一章)

数据仓库概念完整使用 (第一章)

来源：飒榕旅游知识分享网

一、数据仓库概念

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

拿到数据
进行分层建模：目的准确性、易用性、更好的服务与数仓

仓库用来存储公司的所有数据的

1、业务数据

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

2、用户行为数据

用户在使用产品过程中，通过埋点收集与客户端产品交互过程中产生的数据，并发往日志服务器进行保存。比如页面浏览、点击、停留、评论、点赞、收藏等。用户行为数据通常存储在日志文件中。

3、爬虫数据

通常是通过技术手段获取其他公司网站的数据。不建议同学们这样去做。

二、项目需求及架构设计

1、项目需求分析

1）采集平台

（1）用户行为数据采集平台搭建

（2）业务数据采集平台搭建

2）离线需求

3）实时需求

4）思考题

1、项目技术如何选型？
2、框架版本如何选型（Apache、CDH、HDP）
3、服务器使用物理机还是云主机？
4、如何确认集群规模？（假设每台服务器8T硬盘）

2、项目框架

1）技术选型

2）系统数据流程设计

3）框架版本选型

4）服务器选型

5）集群资源规划设计

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

1）生产集群

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

2）测试集群服务器规划

三、用户行为日志

1、用户行为日志概述

用户行为日志的内容，主要包括用户的各项行为信息以及行为所处的环境信息。收集这些信息的主要目的是优化产品和为各项分析统计指标提供数据支撑。收集这些信息的手段通常为埋点。

目前主流的埋点方式，有代码埋点（前端/后端）、可视化埋点、全埋点等。
代码埋点是通过调用埋点SDK函数，在需要埋点的业务逻辑功能位置调用接口，上报埋点数据。例如，我们对页面中的某个按钮埋点后，当这个按钮被点击时，可以在这个按钮对应的 OnClick 函数里面调用SDK提供的数据发送接口，来发送数据。

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

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

2、用户行为日志内容

本项目收集和分析的用户行为信息主要有页面浏览记录、动作记录、曝光记录、启动记录和错误记录。

1、页面浏览记录

页面浏览记录，记录的是访客对页面的浏览行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及页面信息等。

2、动作记录

动作记录，记录的是用户的业务操作行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及动作目标对象信息等。

3、曝光记录

曝光记录，记录的是曝光行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及曝光对象信息等。

4、启动记录

启动记录，记录的是用户启动应用的行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、启动类型及开屏广告信息等。

5、错误记录

启动记录，记录的是用户在使用应用过程中的报错行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、以及可能与报错相关的页面信息、动作信息、曝光信息和动作信息。

4、用户行为日志格式

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

1、页面日志

页面日志，以页面浏览为单位，即一个页面浏览记录，生成一条页面埋点日志。一条完整的页面日志包含，一个页面浏览记录，若干个用户在该页面所做的动作记录，若干个该页面的曝光记录，以及一个在该页面发生的报错记录。除上述行为信息，页面日志还包含了这些行为所处的各种环境信息，包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息等。

{
	"common": {                     -- 环境信息
		"ar": "230000",             -- 地区编码
		"ba": "iPhone",             -- 手机品牌
		"ch": "Appstore",           -- 渠道
		"is_new": "1",              -- 是否首日使用，首次使用的当日，该字段值为1，过了24:00，该字段置为0。
		"md": "iPhone 8",           -- 手机型号
		"mid": "YXfhjAYH6As2z9Iq",  -- 设备id
		"os": "iOS 13.2.9",         -- 操作系统
		"uid": "485",               -- 会员id
		"vc": "v2.1.134"            -- app版本号
	},
	"actions": [{                   -- 动作(事件)
		"action_id": "favor_add",   -- 动作id
		"item": "3",                -- 目标id
		"item_type": "sku_id",      -- 目标类型
		"ts": 1585744376605         -- 动作时间戳
	    }
	],
	"displays": [{                  -- 曝光
			"displayType": "query", -- 曝光类型
			"item": "3",            -- 曝光对象id
			"item_type": "sku_id",  -- 曝光对象类型
			"order": 1,             -- 出现顺序
			"pos_id": 2             -- 曝光位置
		},
		{
			"displayType": "promotion",
			"item": "6",
			"item_type": "sku_id",
			"order": 2,
			"pos_id": 1
		},
		{
			"displayType": "promotion",
			"item": "9",
			"item_type": "sku_id",
			"order": 3,
			"pos_id": 3
		},
		{
			"displayType": "recommend",
			"item": "6",
			"item_type": "sku_id",
			"order": 4,
			"pos_id": 2
		},
		{
			"displayType": "query ",
			"item": "6",
			"item_type": "sku_id",
			"order": 5,
			"pos_id": 1
		}
	],
	"page": {                          -- 页面信息
		"during_time": 78,           -- 持续时间毫秒
		"item": "3", 	               -- 目标id
		"item_type": "sku_id",         -- 目标类型
		"last_page_id": "login",       -- 上页类型
		"page_id": "good_detail",      -- 页面ID
		"sourceType": "promotion"      -- 来源类型
	},                                 
	"err": {                           --错误
		"error_code": "1234",          --错误码
		"msg": "***********"           --错误信息
	},                                 
	"ts": 1585744374423                --跳入时间戳
}

2、启动日志

启动日志以启动为单位，及一次启动行为，生成一条启动日志。一条完整的启动日志包括一个启动记录，一个本次启动时的报错记录，以及启动时所处的环境信息，包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息等。

{
  "common": {
    "ar": "370000",
    "ba": "Honor",
    "ch": "wandoujia",
    "is_new": "1",
    "md": "Honor 20s",
    "mid": "eQF5boERMJFOujcp",
    "os": "Android 11.0",
    "uid": "76",
    "vc": "v2.1.134"
  },
  "start": {   
    "entry": "icon",         --icon手机图标  notice 通知   install 安装后启动
    "loading_time": 18803,  --启动加载时间
    "open_ad_id": 7,        --广告页ID
    "open_ad_ms": 3449,    -- 广告总共播放时间
    "open_ad_skip_ms": 19   --  用户跳过广告时点
  },
"err":{                     --错误
"error_code": "1234",      --错误码
    "msg": "***********"       --错误信息
},
  "ts": 1585744304000
}

3、服务器和JDK准备

1、服务器搭建之前教程有就不搭建了

2、编写集群分发脚本xsync

（1）需求：循环复制文件到所有节点的相同目录下

（2）需求分析

① rsync命令原始拷贝：
rsync  -av     /opt/module  		 root@hadoop103:/opt/

② 期望脚本：
xsync要同步的文件名称

③ 说明：在/home/atguigu/bin这个目录下存放的脚本，atguigu用户可以在系统任何地方直接执行。

（3）脚本实现

① 在用的家目录/home/atguigu下创建bin文件夹
[atguigu@hadoop102 ~]$ mkdir bin

② 在/home/atguigu/bin目录下创建xsync文件，以便全局调用
[atguigu@hadoop102 ~]$ cd /home/atguigu/bin
[atguigu@hadoop102 ~]$ vim xsync

在该文件中编写如下代码

#!/bin/bash
#1. 判断参数个数
if [ $# -lt 1 ]
then
  echo Not Enough Arguement!
  exit;
fi
#2. 遍历集群所有机器
for host in hadoop102 hadoop103 hadoop104
do
  echo ====================  $host  ====================
  #3. 遍历所有目录，挨个发送
  for file in $@
  do
    #4 判断文件是否存在
    if [ -e $file ]
    then
      #5. 获取父目录
      pdir=$(cd -P $(dirname $file); pwd)
      #6. 获取当前文件的名称
      fname=$(basename $file)
      ssh $host "mkdir -p $pdir"
      rsync -av $pdir/$fname $host:$pdir
    else
      echo $file does not exists!
    fi
  done
done

③ 修改脚本xsync具有执行权限
[atguigu@hadoop102 bin]$ chmod +x xsync

④ 测试脚本
[atguigu@hadoop102 bin]$ xsync xsync

5、无密登录配置

(1) hadoop102上生成公钥和私钥

[atguigu@hadoop102 .ssh]$ ssh-keygen -t rsa

然后敲（三个回车），就会生成两个文件id_rsa（私钥）、id_rsa.pub（公钥）

(2) 将hadoop102公钥拷贝到要免密登录的目标机器上

[atguigu@hadoop102 .ssh]$ ssh-copy-id hadoop102
[atguigu@hadoop102 .ssh]$ ssh-copy-id hadoop103
[atguigu@hadoop102 .ssh]$ ssh-copy-id hadoop104

(3) hadoop103上生成公钥和私钥

[atguigu@hadoop103 .ssh]$ ssh-keygen -t rsa

然后敲（三个回车），就会生成两个文件id_rsa（私钥）、id_rsa.pub（公钥）

(4) 将hadoop103公钥拷贝到要免密登录的目标机器上

[atguigu@hadoop103 .ssh]$ ssh-copy-id hadoop102
[atguigu@hadoop103 .ssh]$ ssh-copy-id hadoop103
[atguigu@hadoop103 .ssh]$ ssh-copy-id hadoop104

(5) JDK准备

1）卸载现有JDK（3台节点）

[atguigu@hadoop102 opt]# sudo rpm -qa | grep -i java | xargs -n1 sudo rpm -e --nodeps
[atguigu@hadoop103 opt]# sudo rpm -qa | grep -i java | xargs -n1 sudo rpm -e --nodeps
[atguigu@hadoop104 opt]# sudo rpm -qa | grep -i java | xargs -n1 sudo rpm -e --nodeps

（1）rpm -qa：表示查询所有已经安装的软件包
（2）grep -i：表示过滤时不区分大小写
（3）xargs -n1：表示一次获取上次执行结果的一个值
（4）rpm -e --nodeps：表示卸载软件

2）用XShell工具将JDK导入到hadoop102的/opt/software文件夹下面

3）在Linux系统下的opt目录中查看软件包是否导入成功

[atguigu@hadoop102 software]# ls /opt/software/

看到如下结果：
jdk-8u212-linux-x.tar.gz

4）解压JDK到/opt/module目录下

[atguigu@hadoop102 software]# tar -zxvf jdk-8u212-linux-x.tar.gz -C /opt/module/
[atguigu@hadoop102 module]$ mv jdk1.8.0_212/ jdk

5）配置JDK环境变量

（1）新建/etc/profile.d/my_env.sh文件
[atguigu@hadoop102 module]# sudo vim /etc/profile.d/my_env.sh

添加如下内容，然后保存（:wq）退出
#JAVA_HOME
export JAVA_HOME=/opt/module/jdk
export PATH=$PATH:$JAVA_HOME/bin

（2）让环境变量生效
[atguigu@hadoop102 software]$ source /etc/profile.d/my_env.sh

6）测试JDK是否安装成功

[atguigu@hadoop102 module]# java -version
如果能看到以下结果、则Java正常安装
java version "1.8.0_212"

7）分发JDK

[atguigu@hadoop102 module]$ xsync /opt/module/jdk

8）分发环境变量配置文件

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

9）分别在hadoop103、hadoop104上执行source

[atguigu@hadoop103 module]$ source /etc/profile.d/my_env.sh
[atguigu@hadoop104 module]$ source /etc/profile.d/my_env.sh

6、环境变量配置说明

Linux的环境变量可在多个文件中配置，如/etc/profile，/etc/profile.d/*.sh，_/.bashrc，/.bash_profile等，下面说明上述几个文件之间的关系和区别。
bash的运行模式可分为login shell和non-login shell。
例如，我们通过终端，输入用户名、密码，登录系统之后，得到就是一个login shell。而当我们执行以下命令ssh hadoop103 command，在hadoop103执行command的就是一个non-login shell。

这两种shell的主要区别在于，它们启动时会加载不同的配置文件，login shell启动时会加载/etc/profile，_{/.bash_profile，}/.bashrc。non-login shell启动时会加载~/.bashrc。
而在加载_{/.bashrc（实际是}/.bashrc中加载的/etc/bashrc）或/etc/profile时，都会执行如下代码片段，

因此不管是login shell还是non-login shell，启动时都会加载/etc/profile.d/*.sh中的环境变量。

7、模拟数据

1、使用说明

1）将模拟数据文件上传到hadoop102的/opt/module/applog目录下

（1）创建applog路径
[atguigu@hadoop102 module]$ mkdir /opt/module/applog
（2）上传文件到/opt/module/applog目录

2）配置文件

（1）application.yml文件
可以根据需求生成对应日期的用户行为日志。
[atguigu@hadoop102 applog]$ vim application.yml

修改如下内容

# 外部配置打开
logging.config: "./logback.xml"
#业务日期  注意：并不是Linux系统生成日志的日期，而是生成数据中的时间
mock.date: "2020-06-14"

#模拟数据发送模式
#mock.type: "http"
#mock.type: "kafka"
mock.type: "log"

#http模式下，发送的地址
mock.url: "http://hdp1/applog"

#kafka模式下，发送的地址
mock:
  kafka-server: "hdp1:9092,hdp2:9092,hdp3: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.atgugu.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>

3）生成日志

（1）进入到/opt/module/applog路径，执行以下命令
[atguigu@hadoop102 applog]$ java -jar gmall2020-mock-log-2021-10-10.jar
（2）在/opt/module/applog/log目录下查看生成日志
[atguigu@hadoop102 log]$ ll

2、集群日志生成脚本

[atguigu@hadoop102 ~]$ echo $PATH
/usr/local/bin:/usr/bin:/usr/local/sbin:/usr/sbin:/home/atguigu/.local/bin:/home/atguigu/bin

（1）在/home/atguigu/bin目录下创建脚本lg.sh

[atguigu@hadoop102 bin]$ vim lg.sh

（2）在脚本中编写如下内容

#!/bin/bash
for i in hadoop102 hadoop103; do
    echo "========== $i =========="
    ssh $i "cd /opt/module/applog/; java -jar gmall2020-mock-log-2021-10-10.jar >/dev/null 2>&1 &"
done

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

（3）修改脚本执行权限

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

（4）将jar包及配置文件上传至hadoop103的/opt/module/applog/路径

（5）启动脚本

[atguigu@hadoop102 module]$ lg.sh

（6）分别在hadoop102、hadoop103的/opt/module/applog/log目录上查看生成的数据

[atguigu@hadoop102 logs]$ ls
app.2020-06-14.log
[atguigu@hadoop103 logs]$ ls
app.2020-06-14.log

四、用户行为数据采集模块

1、数据通道

2、环境准备

(1) 集群所有进程查看脚本

1）在/home/atguigu/bin目录下创建脚本xcall

[atguigu@hadoop102 bin]$ vim xcall

2）在脚本中编写如下内容

#! /bin/bash
 
for i in hadoop102 hadoop103 hadoop104
do
    echo --------- $i ----------
    ssh $i "$*"
done

3）修改脚本执行权限

[atguigu@hadoop102 bin]$ chmod 777 xcall

4）启动脚本

[atguigu@hadoop102 bin]$ xcall.sh jps

(2) Hadoop安装

1) 安装步骤看之前的

2) 项目经验 (可看可不看)

（1）项目经验之HDFS存储多目录

HDFS的DataNode节点保存数据的路径由dfs.datanode.data.dir参数决定，其默认值为file://${hadoop.tmp.dir}/dfs/data，若服务器有多个磁盘，必须对该参数进行修改。如服务器磁盘如上图所示，则该参数应修改为如下的值。

<property>
    <name>dfs.datanode.data.dir</name>
    <value>file:///dfs/data1,file:///hd2/dfs/data2,file:///hd3/dfs/data3,file:///hd4/dfs/data4</value>
</property>

（2）项目经验之集群数据均衡

①节点间数据均衡

开启数据均衡命令：

start-balancer.sh -threshold 10

对于参数10，代表的是集群中各个节点的磁盘空间利用率相差不超过10%，可根据实际情况进行调整。

停止数据均衡命令：

stop-balancer.sh

②磁盘间数据均衡
生成均衡计划（我们只有一块磁盘，不会生成计划）

hdfs diskbalancer -plan hadoop103

执行均衡计划

hdfs diskbalancer -execute hadoop103.plan.json

查看当前均衡任务的执行情况

hdfs diskbalancer -query hadoop103

取消均衡任务

hdfs diskbalancer -cancel hadoop103.plan.json

（3）项目经验之Hadoop参数调优

（4）DataNode向NameNode上报

NameNode就得有线程来接收上报的请求、线程数设置多少合适、太多太浪费、可以使用下面公式计算

① HDFS参数调优hdfs-site.xml

The number of Namenode RPC server threads that listen to requests from clients. If dfs.namenode.servicerpc-address is not configured then Namenode RPC server threads listen to requests from all nodes.
NameNode有一个工作线程池，用来处理不同DataNode的并发心跳以及客户端并发的元数据操作。
对于大集群或者有大量客户端的集群来说，通常需要增大参数dfs.namenode.handler.count的默认值10。

dfs.namenode.handler.count的默认值10。
<property>
    <name>dfs.namenode.handler.count</name>
    <value>10</value>
</property>

dfs.namenode.handler.count=，比如集群规模为8台时，此参数设置为41。可通过简单的python代码计算该值，代码如下。

[atguigu@hadoop102 ~]$ python
Python 2.7.5 (default, Apr 11 2018, 07:36:10) 
[GCC 4.8.5 20150623 (Red Hat 4.8.5-28)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import math
>>> print int(20*math.log(8))
41
>>> quit()

② YARN参数调优yarn-site.xml

情景描述：总共7台机器，每天几亿条数据，数据源->Flume->Kafka->HDFS->Hive

面临问题：数据统计主要用HiveSQL，没有数据倾斜，小文件已经做了合并处理，开启的JVM重用，而且IO没有阻塞，内存用了不到50%。但是还是跑的非常慢，而且数据量洪峰过来时，整个集群都会宕掉。基于这种情况有没有优化方案。

解决办法：
内存利用率不够。这个一般是Yarn的2个配置造成的，单个任务可以申请的最大内存大小，和Hadoop单个节点可用内存大小。调节这两个参数能提高系统内存的利用率。
（a）yarn.nodemanager.resource.memory-mb
表示该节点上YARN可使用的物理内存总量，默认是8192（MB），注意，如果你的节点内存资源不够8GB，则需要调减小这个值，而YARN不会智能的探测节点的物理内存总量。
（b）yarn.scheduler.maximum-allocation-mb
单个任务可申请的最多物理内存量，默认是8192（MB）。

3) Zookeeper安装

1) 安装步骤

查看我之前的

4) Kafka安装

1) 安装步骤

查看我之前的

5) Flume安装

1) 安装步骤

按照采集通道规划，需在hadoop102，hadoop103，hadoop104三台节点分别部署一个Flume。可参照以下步骤先在hadoop102安装，然后再进行分发。

2) 分发Flum

atguigu@hadoop102 ~]$ xsync /opt/module/flume/

3) 项目经验

（1）堆内存调整

Flume堆内存通常设置为4G或更高，配置方式如下：
修改/opt/module/flume/conf/flume-env.sh文件，配置如下参数（虚拟机环境暂不配置）

export JAVA_OPTS="-Xms4096m -Xmx4096m -Dcom.sun.management.jmxremote"

注：
-Xms表示JVM Heap(堆内存)最小尺寸，初始分配；
-Xmx 表示JVM Heap(堆内存)最大允许的尺寸，按需分配。

3、日志采集Flume

1、日志采集Flume配置概述

按照规划，需要采集的用户行为日志文件分布在hadoop102，hadoop103两台日志服务器，故需要在hadoop102，hadoop103两台节点配置日志采集Flume。日志采集Flume需要采集日志文件内容，并对日志格式（JSON）进行校验，然后将校验通过的日志发送到Kafka。

此处可选择TaildirSource和KafkaChannel，并配置日志校验。
选择TailDirSource和KafkaChannel的原因如下：

1）介绍

Source
	taildirSource  读取文件的数据、实时的读取文件的数据、支持断点续传-用得较多
	avroSource  在flume之间互相传输的Source 一般配合avroSink 使用-其次
	ncSource 接收网络端口
	execSource 和taildirSource一样也是实时监控、不支持断点续传-一般没人用
	spoolingSource 支持断点续传、监控的不是文件、监控的文件夹 一般没人用
    kafkaSource 就是一个kafka的一个消费者
channel	
	fileChannel 慢、小优化、增加索引、从内存里找到指定位置读写、加速查询、如果内存挂了、他还会把索引被备份到磁盘上-如果内存-磁盘都挂了、还还会进行二次备份
	memoryChannel
	kafkaChannel
sink	
	hdfs sink
	kafka sink 给kafka topic发数据-生产者
	avro

2）TailDirSource

TailDirSource相比ExecSource、SpoolingDirectorySource的优势

TailDirSource：断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置，实现断点续传。

ExecSource可以实时搜集数据，但是在Flume不运行或者Shell命令出错的情况下，数据将会丢失。

SpoolingDirectorySource监控目录，支持断点续传。

3）KafkaChannel

采用Kafka Channel，省去了Sink，提高了效率。
日志采集Flume关键配置如下：

2、日志采集Flume配置实操

1）创建Flume配置文件

[atguigu@hadoop104 flume]$ mkdir job
[atguigu@hadoop104 flume]$ vim job/file_to_kafka.conf

2）配置文件内容如下

红色部分切换自己

#定义组件
a1.sources = r1
a1.channels = c1
#配置source
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
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.gmall.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
#组装
a1.sources.r1.channels = c1

3）编写Flume

（1）创建Maven工程flume-interceptor
（2）创建包：com.atguigu.gmall.flume.interceptor
（3）在pom.xml文件中添加如下配置

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

（4）在com.atguigu.gmall.flume.utils包下创建JSONUtil类

package com.atguigu.gmall.flume.utils;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONException;

public class JSONUtil {
/*
* 通过异常判断是否是json字符串
* 是：返回true  不是：返回false
* */
    public static boolean isJSONValidate(String log){
        try {
            JSONObject.parseObject(log);
            return true;
        }catch (JSONException e){
            return false;
        }
    }
}

（5）在com.atguigu.gmall.flume.interceptor包下创建ETLInterceptor类

package com.atguigu.gmall.flume.interceptor;

import com.atguigu.gmall.flume.utils.JSONUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;


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

public class ETLInterceptor implements Interceptor {

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
		
		//1、获取body当中的数据并转成字符串
        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);
		//2、判断字符串是否是一个合法的json，是：返回当前event；不是：返回null
        if (JSONUtil.isJSONValidate(log)) {
            return event;
        } else {
            return null;
        }
    }

    @Override
    public 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{

        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }
        @Override
        public void configure(Context context) {

        }

    }

    @Override
    public void close() {

    }
}

（6）打包

（7）需要先将打好的包放入到hadoop102的/opt/module/flume/lib文件夹下面

3、日志采集Flume测试

1）启动Zookeeper、Kafka集群

2）启动hadoop102的日志采集Flume

[atguigu@hadoop102 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/file_to_kafka.conf -Dflume.root.logger=info,console

3）启动一个Kafka的Console-Consumer

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

4）生成模拟数据

[atguigu@hadoop102 ~]$ lg.sh

5）观察Kafka消费者是否能消费到数据

4、日志采集Flume启停脚本

1）分发日志采集Flume配置文件和

若上述测试通过，需将hadoop102节点的Flume的配置文件和jar包，向另一台日志服务器发送一份。

[atguigu@hadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/
[atguigu@hadoop102 flume]$ scp lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar hadoop103:/opt/module/flume/lib/

2）方便起见，此处编写一个日志采集Flume进程的启停脚本

[atguigu@hadoop102 bin]$ vim f1.sh

在脚本中填写如下内容

#!/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 -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"
        done
};; 
"stop"){
        for i in hadoop102 hadoop103
        do
                echo " --------停止 $i 采集flume-------"
                ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk  '{print \$2}' | xargs -n1 kill -9 "
        done

};;
esac

3）增加脚本执行权限

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

4）f1启动

[atguigu@hadoop102 module]$ f1.sh start

5）f2停止

[atguigu@hadoop102 module]$ f1.sh stop

因篇幅问题不能全部显示，请点此查看更多更全内容

查看全文