第一章 数据仓库概念

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

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

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

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

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

第二章 项目需求及架构设计

2.1 项目需求分析

2.2 项目框架

2.2.1 技术选型

2.2.2 系统数据流程设计

2.2.3 框架版本选型

2.2.4服务器选型

2.2.5 集群规模

2.2.6 集群资源规划设计

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

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

1	2	3	4	5	6	7	8	9	10
nn	nn	dn	dn	dn	dn	dn	dn	dn	dn
		rm	rm	nm	nm	nm	nm	nm	nm
		nm	nm
							zk	zk	zk
							kafka	kafka	kafka
							Flume	Flume	flume
		Hbase	Hbase	Hbase
hive	hive
mysql	mysql
spark	spark
Azkaban	Azkaban				ES	ES

2）测试集群服务器规划

服务名称	子服务	服务器 hadoop102	服务器 hadoop103	服务器 hadoop104
HDFS	NameNode	√
	DataNode	√	√	√
	SecondaryNameNode			√
Yarn	NodeManager	√	√	√
	Resourcemanager		√
Zookeeper	Zookeeper Server	√	√	√
Flume（采集日志）	Flume	√	√
Kafka	Kafka	√	√	√
Flume（消费Kafka）	Flume			√
Hive	Hive	√
MySQL	MySQL	√
Sqoop	Sqoop	√
Presto	Coordinator	√
	Worker		√	√
Azkaban	AzkabanWebServer	√
	AzkabanExecutorServer	√
Spark		√
Kylin		√
HBase	HMaster	√
	HRegionServer	√	√	√
Superset		√
Atlas		√
Solr	Jar	√
服务数总计		19	8	8

第三章 数据生成模块

3.1 目标数据

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

3.1.1 页面

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

字段名称	字段描述
page_id	页面id home（"首页"）, category（"分类页"）, discovery（"发现页"）, top_n（"热门排行"）, favor（"收藏页"）, search（"搜索页"）, good_list（"商品列表页"）, good_detail（"商品详情"）, good_spec（"商品规格"）, comment（"评价"）, comment_done（"评价完成"）, comment_list（"评价列表"）, cart（"购物车"）, trade（"下单结算"）, payment（"支付页面"）, payment_done（"支付完成"）, orders_all（"全部订单"）, orders_unpaid（"订单待支付"）, orders_undelivered（"订单待发货"）, orders_unreceipted（"订单待收货"）, orders_wait_comment（"订单待评价"）, mine（"我的"）, activity（"活动"）, login（"登录"）, register（"注册"）;
last_page_id	上页id
page_item_type	页面对象类型 sku_id（"商品skuId"）, keyword（"搜索关键词"）, sku_ids（"多个商品skuId"）, activity_id（"活动id"）, coupon_id（"购物券id"）;
page_item	页面对象id
sourceType	页面来源类型 promotion（"商品推广"）, recommend（"算法推荐商品"）, query（"查询结果商品"）, activity（"促销活动"）;
during_time	停留时间（毫秒）
ts	跳入时间

3.1.2 事件

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

字段名称	字段描述
action_id	动作id favor_add（"添加收藏"）, favor_canel（"取消收藏"）, cart_add（"添加购物车"）, cart_remove（"删除购物车"）, cart_add_num（"增加购物车商品数量"）, cart_minus_num（"减少购物车商品数量"）, trade_add_address（"增加收货地址"）, get_coupon（"领取优惠券"）; 注：对于下单、支付等业务数据，可从业务数据库获取。
item_type	动作目标类型 sku_id（"商品"）, coupon_id（"购物券"）;
item	动作目标id
ts	动作时间

3.1.3 曝光

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

字段名称	字段描述
displayType	曝光类型 promotion（"商品推广"）, recommend（"算法推荐商品"）, query（"查询结果商品"）, activity（"促销活动"）;
item_type	曝光对象类型 sku_id（"商品skuId"）, activity_id（"活动id"）;
item	曝光对象id
order	曝光顺序

3.1.4 启动

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

字段名称	字段描述
entry	启动入口 icon（"图标"）, notification（"通知"）, install（"安装后启动"）;
loading_time	启动加载时间
open_ad_id	开屏广告id
open_ad_ms	广告播放时间
open_ad_skip_ms	用户跳过广告时间
ts	启动时间

3.1.5 错误

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

字段名称	字段描述
error_code	错误码
msg	错误信息

3.2 数据埋点

3.2.1 主流埋点方式（了解）

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

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

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

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

3.2.2 埋点数据上报时机

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

方式一，在离开该页面时，上传在这个页面产生的所有数据（页面、事件、曝光、错误等）。优点，批处理，减少了服务器接收数据压力。缺点，不是特别及时。

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

本次项目采用方式一埋点。

3.2.3 埋点数据日志结构

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

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

（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": 7648,        -- 持续时间毫秒
    "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": 1989   --  用户跳过广告时点
  },
"err":{                     --错误
"error_code": "1234",      --错误码
    "msg": "***********"       --错误信息
},
  "ts": 1585744304000
}

3.3 服务器和JDK准备

分别安装hadoop102、hadoop103、hadoop104三台主机。

相关内容参考：

3.4 模拟数据

3.4.1 使用说明

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

（1）创建applog路径

[atguigu@hadoop102 module]$ mkdir /opt/module/applog

（2）上传文件application.yml到/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%npattern>
        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}.logfileNamePattern>
        rollingPolicy>
        <encoder>
            <pattern>%msg%npattern>
        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-01-22.jar

（2）在/opt/module/applog/log目录下查看生成日志

[atguigu@hadoop102 log]$ ll

3.4.2 集群日志生成脚本

    在hadoop102的/home/atguigu目录下创建bin目录，这样脚本可以在服务器的任何目录执行。

[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-01-22.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

这里意思是 1给黑洞，2给1

（3）修改脚本执行权限

[atguigu@hadoop102 bin]$ chmod u+x 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.2021-01-22.log
[atguigu@hadoop103 logs]$ ls
app.2020-01-22.log

第四章 数据采集模块

相关内容参考：

4.1 Hadoop 安装

集群规划

	服务器hadoop102	服务器hadoop103	服务器hadoop104
HDFS	NameNode DataNode	DataNode	DataNode SecondaryNameNode
Yarn	NodeManager	Resourcemanager NodeManager	NodeManager

注意：尽量使用离线方式安装
安装过程参见：

4.2 Zookeeper 安装

集群规划

	服务器hadoop102	服务器hadoop103	服务器hadoop104
Zookeeper	Zookeeper	Zookeeper	Zookeeper

安装过程参见：

4.3 Kafka 安装

集群规划

	服务器hadoop102	服务器hadoop103	服务器hadoop104
Kafka	Kafka	Kafka	Kafka

安装过程参见：

4.3.1 创建Kafka Topic

进入到/opt/module/kafka/目录下创建日志主题
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181  --create --replication-factor 1 --partitions 1 --topic topic_log

4.4 采集日志Flume

4.4.1 Flume安装

	服务器hadoop102	服务器hadoop103	服务器hadoop104
Flume(采集日志)	Flume	Flume

安装过程参见：

4.4.2 Flume组件选型

1）Source

（1）Taildir Source相比Exec Source、Spooling Directory Source的优势

TailDir Source：断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置，实现断点续传。不会丢数据，但是有可能会导致数据重复。

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

Spooling Directory Source监控目录，支持断点续传。

（2）batchSize大小如何设置？

答：Event 1K左右时，500-1000合适（默认为100）

2）Channel
采用Kafka Channel

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

注意在Flume1.7以前，Kafka Channel很少有人使用，因为发现parseAsFlumeEvent这个配置起不了作用。也就是无论parseAsFlumeEvent配置为true还是false，都会转为Flume Event。这样的话，造成的结果是，会始终都把Flume的headers中的信息混合着内容一起写入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文件

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

在文件配置如下内容

#为各组件命名
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.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

#绑定source和channel以及sink和channel的关系
a1.sources.r1.channels = c1

    注意：com.atguigu.flume.interceptor.ETLInterceptor是自定义的拦截器的全类名。需要根据用户自定义的拦截器做相应修改。

4.4.4 Flume拦截器

1）创建Maven工程flume-interceptor

2）创建包名：com.atguigu.flume.interceptor

3）在pom.xml文件中添加如下配置

<dependencies>
    <dependency>
        <groupId>org.apache.flumegroupId>
        <artifactId>flume-ng-coreartifactId>
        <version>1.9.0version>
        <scope>providedscope>
    dependency>

    <dependency>
        <groupId>com.alibabagroupId>
        <artifactId>fastjsonartifactId>
        <version>1.2.62version>
    dependency>
dependencies>

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-pluginartifactId>
            <version>2.3.2version>
            <configuration>
                <source>1.8source>
                <target>1.8target>
            configuration>
        plugin>
        <plugin>
            <artifactId>maven-assembly-pluginartifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependenciesdescriptorRef>
                descriptorRefs>
            configuration>
            <executions>
                <execution>
                    <id>make-assemblyid>
                    <phase>packagephase>
                    <goals>
                        <goal>singlegoal>
                    goals>
                execution>
            executions>
        plugin>
    plugins>
build>

    注意：scope中provided的含义是编译时用该jar包。打包时时不用。因为集群上已经存在flume的jar包。只是本地编译时用一下。

4）在com.atguigu.flume.interceptor包下创建JSONUtils类

package com.atguigu.flume.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;

public class JSONUtils {
    public static boolean isJSONValidate(String log){
        try {
            JSON.parse(log);
            return true;
        }catch (JSONException e){
            return false;
        }
    }
}

5）在com.atguigu.flume.interceptor包下创建LogInterceptor类

package com.atguigu.flume.interceptor;

import com.alibaba.fastjson.JSON;
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) {

        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);

        if (JSONUtils.isJSONValidate(log)) {
            return event;
        } else {
            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）需要先将打好的包放入到hadoop102的/opt/module/flume/lib文件夹下面。

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

8）分发Flume到hadoop103、hadoop104

[atguigu@hadoop102 module]$ xsync flume/

9）分别在hadoop102、hadoop103上启动Flume

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

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

4.4.5 测试Flume-Kafka通道

（1）生成日志

[atguigu@hadoop102 ~]$ lg.sh

（2）消费Kafka数据，观察控制台是否有数据获取到

[atguigu@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）在/home/atguigu/bin目录下创建脚本f1.sh

[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 --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 "ps -ef | grep file-flume-kafka | grep -v grep |awk  '{print \$2}' | xargs -n1 kill -9 "
        done

};;
esac

说明1：nohup，该命令可以在你退出帐户/关闭终端之后继续运行相应的进程。nohup就是不挂起的意思，不挂断地运行命令。
说明2：awk 默认分隔符为空格
说明3：$2是在“”双引号内部会被解析为脚本的第二个参数，但是这里面想表达的含义是awk的第二个值，所以需要将他转义，用\$2表示。
说明4：xargs 表示取出前面命令运行的结果，作为后面命令的输入参数。
（2）增加脚本执行权限

[atguigu@hadoop102 bin]$ chmod u+x f1.sh

（3）f1集群启动脚本

[atguigu@hadoop102 module]$ f1.sh start

（4）f1集群停止脚本

[atguigu@hadoop102 module]$ f1.sh stop

4.5 消费Kafka数据Flume

集群规划

	服务器hadoop102	服务器hadoop103	服务器hadoop104
Flume（消费Kafka）			Flume

4.5.1 项目经验之Flume组件选型

1）FileChannel和MemoryChannel区别

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

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

选型：

金融类公司、对钱要求非常准确的公司通常会选择FileChannel

传输的是普通日志信息（京东内部一天丢100万-200万条，这是非常正常的），通常选择MemoryChannel。

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恢复数据。

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几个参数综合作用，效果如下：

①文件在达到128M时会滚动生成新文件
②文件创建超3600秒时会滚动生成新文件

4.5.2 消费者Flume配置

1）Flume配置分析

加上时间戳拦截器，用来解决零点漂移问题。

为什么会发生零点漂移？

比如2022-01-24 23:59:59生成的日志文件，然后数据经过第一层的flume采集，加上kafka的缓冲，然后到 集群的另一台上的第二层的flume的时候，时间肯定就会到2022-01-25 00:00:XX了，这样一来，如果采用当前系统时间作为timestamp的话，2022-01-24 的日志数据就会上传到hdfs上的2022-01-25 的目录下。因为Kafka Source会为其加上该header，value为当前系统的时间戳Kafka Source会为其加上该header，value为当前系统的时间戳

所以我们需要再第二层flume里面写这么一个加时间戳的拦截器，把日志文件里面的时间添加到event的header里面

2）Flume的具体配置如下：
    （1）在hadoop104的/opt/module/flume/conf目录下创建kafka-flume-hdfs.conf文件

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

在文件配置如下内容

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

## source1
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.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.flume.interceptor.TimeStampInterceptor$Builder

## channel1
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/behavior1
a1.channels.c1.dataDirs = /opt/module/flume/data/behavior1/


## sink1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = log-
a1.sinks.k1.hdfs.round = false

#控制生成的小文件
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.3 Flume时间戳拦截器

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

解决的思路：拦截json日志，通过fastjson框架解析json，获取实际时间ts。将获取的ts时间写入拦截器header头，header的key必须是timestamp，因为Flume框架会根据这个key的值识别为时间，写入到HDFS。
1）在com.atguigu.flume.interceptor包下创建TimeStampInterceptor类

package com.atguigu.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;
    }

    @Override
    public void close() {
    }
    public static class Builder implements Interceptor.Builder{
        @Override
        public Interceptor build() {
            return new TimeStampInterceptor();
        }
        @Override
        public void configure(Context context) {
        }
    }
}

2）重新打包
3）需要先将打好的包放入到hadoop102的/opt/module/flume/lib文件夹下面。

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

4）分发Flume到hadoop103、hadoop104

[atguigu@hadoop102 module]$ xsync flume/

4.5.4 消费者Flume启动停    

在脚本中填写如下内容

#! /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 "ps -ef | grep kafka-flume-hdfs | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
        done

};;
esac

（2）增加脚本执行权限

[atguigu@hadoop102 bin]$ chmod u+x f2.sh

（3）f2集群启动脚本

[atguigu@hadoop102 module]$ f2.sh start

（4）f2集群停止脚本

[atguigu@hadoop102 module]$ f2.sh stop

4.3.5 项目经验之Flume内存优化

1）问题描述：如果启动消费Flume抛出如下异常

ERROR hdfs.HDFSEventSink: process failed
java.lang.OutOfMemoryError: GC overhead limit exceeded

2）解决方案步骤
（1）在hadoop102服务器的/opt/module/flume/conf/flume-env.sh文件中增加如下配置

export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"

（2）同步配置到hadoop103、hadoop104服务器

[atguigu@hadoop102 conf]$ xsync flume-env.sh

3）Flume内存参数设置及优化

JVM heap一般设置为4G或更高

-Xmx与-Xms最好设置一致，减少内存抖动带来的性能影响，如果设置不一致容易导致频繁fullgc。

-Xms表示JVM Heap（堆内存）最小尺寸，初始分配；-Xmx 表示JVM Heap(堆内存)最大允许的尺寸，按需分配。如果不设置一致，容易在初始化时，由于内存不够，频繁触发fullgc。

4.6 采集通道启动/停止脚本

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

[atguigu@hadoop102 bin]$ vim cluster.sh

    在脚本中填写如下内容

#!/bin/bash

case $1 in
"start"){
        echo ================== 启动 集群 ==================

        #启动 Zookeeper集群
        zk.sh start

        #启动 Hadoop集群
        hdp.sh start

        #启动 Kafka采集集群
        kf.sh start

        #启动 Flume采集集群
        f1.sh start

        #启动 Flume消费集群
        f2.sh start

        };;
"stop"){
        echo ================== 停止 集群 ==================

        #停止 Flume消费集群
        f2.sh stop

        #停止 Flume采集集群
        f1.sh stop

        #停止 Kafka采集集群
        kf.sh stop

        #停止 Hadoop集群
        hdp.sh stop

        #停止 Zookeeper集群
        zk.sh stop

};;
esac

（2）增加脚本执行权限

[atguigu@hadoop102 bin]$ chmod u+x cluster.sh    

（3）cluster集群启动脚本

[atguigu@hadoop102 module]$ cluster.sh start

（4）cluster集群停止脚本

[atguigu@hadoop102 module]$ cluster.sh stop

 产生日志

[atguigu@hadoop102 module]$ lg.sh

查看HDFS