Flume 从0到高手一站式养成记

心已赠人 2022-09-03 13:15 169阅读 0赞

### 文章目录 ###

*  一、学前必备知识
 *  二、极速入门 Flume
 *  三、极速上手 Flume 使用
 *   *  3.1 案例：Flume 的 Hello World!
     *  3.2 案例：采集文件内容上传至 HDFS
     *  3.3 案例：采集网站日志上传至 HDFS
 *  四、Flume 高级组件
 *   *  4.1 Source Interceptors
     *  4.2 Channel Selectors
     *  4.3 Sink Processors
     *  4.4 Failover Sink Processor
 *  五、Flume 出神入化篇

# 一、学前必备知识 #

1.  [2021年 全网最细大数据学习笔记（一）：初识 Hadoop][2021_ _ Hadoop]
2.  [2021年 全网最细大数据学习笔记（二）：Hadoop 伪分布式安装][2021_ _Hadoop]
3.  [2021年 全网最细大数据学习笔记（三）：Hadoop 集群的搭建与配置][2021_ _Hadoop 1]
4.  [2021年 全网最细大数据学习笔记（四）：Hadoop 之 HDFS的基本使用][2021_ _Hadoop _ HDFS]
5.  [2021年 全网最细大数据学习笔记（五）：Zookeeper 集群][2021_ _Zookeeper]
6.  [2021年 全网最细大数据学习笔记（六）：Hadoop 之 HDFS 进程详解][2021_ _Hadoop _ HDFS 1]

# 二、极速入门 Flume #

**什么是 Flume？** Flume 是一个分布式的、高可靠的、高可用的，将大批量的不同数据源的日志数据收集、聚合、移动到 **数据中心(HDFS)** 进行存储的系统，即是 **日志采集和汇总的工具**。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70]  
使用 Flume 采集数据不需要写一行代码，注意是一行代码都不需要，只需要在配置文件中随便写几行配置 Flume 就会死心塌地的给你干活了，是不是很香? **它是目前大数据领域数据采集最常用的一个框架。**

**Flume 的版本变化。** Flume 最初始的发行版本被称为 Flume OG(original generation)，属于 Cloudera 公司。但随着 Flume 功能的扩展，Flume OG 代码量臃肿、核心组件设计不合理、核心配置不标准等缺点都暴露了出来，尤其是 Flume OG 的最后一个发行版本 0.9.4 中，日志传输不稳定的现象尤为严重。为了解决这些问题，2011年10月22号，Cloudera 进行了 Flume 新版本的发布，对 Flume 进行了里程碑式的改动：重构核心组件、核心配置以及代码架构，重构后的版本统称为 Flume NG(Next Generation)；改名的另一原因是 Flume 纳入了 Apache 公司旗下，Cloudera Flume 改名为 Apache Flume。综上所述，Flume 在 1.x 之前的版本称为 Flume OG，在 1.x 之后的版本称为 Flume NG，这两个版本之间有着较大的架构调整，本章主要介绍的是 Flume NG。

**Flume 最基本的结构。** Agent 是 Flume 的运行核心，外部数据进入到 Agent 中，通过 Agent 数据流向另外的地方，下图所示就是一个 Flume 的最基本的结构，只有一个 Agent。  
![在这里插入图片描述][20210702003423927.png]  
**Flume 中概念解析：**

(1) Web Server：外部的原始数据，可以是不同渠道的日志数据。

(2) Event：事件，是流经 Flume 的一个数据单元，由消息头、消息体和消息信息组成，其中消息信息就是 Flume 收集到的日志记录，因为原始数据不方便统一处理，所以需要将原始数据转换成事件。在文本文件中，通常一行记录就是一个 Event，Event 由 Header 和 Body 两部分组成，Header 用来存放该 Event 的一些属性，为 K-V 结构，Body 用来存放该条数据，形式为字节数组：  
![在这里插入图片描述][d060aef43e8d4316babd2704b3f45dd9.png]  
即我们可以在 Source 中给每一条数据的 header 中增加 key-value，在 Channel 和 Sink 中使用 header 中的值了。

(3) Agent：代理，是一个进程承载从外部源事件流到下一个目的地的过程，即一个独立的 Flume 进程，管理 Event 从一个外部源流向外部目标的各组件，组件包含 Source、Channel、Sink。**Agent 是 Flume 系统的核心角色，Agent 本身是一个 Java 进程，一般运行在日志收集节点。**

(4) Source：数据收集组件，消耗外部传递给它的数据。从外部数据源处收集数据，将接收的数据以 Flume 的 Event 格式传递给一个或者多个 Channel。Flume 提供了各种 Source 的实现，包括 Avro Source、**Exce Source**、**Spooling Directory Source**、**NetCat Source**、Syslog Source、Syslog TCP Source、Syslog UDP Source、HTTP Source、HDFS Source、**Kafka Source** 等。如果内置的 Source 无法满足需要， Flume 还支持自定义 Source。

Exec Source：实现文件监控，可以实时监控文件中的新增内容，类似于 linux 中的 tail -f 效果。tail -F 和 tail -f 的区别如下：

1.  tail -F。等同于 -follow=name --retry，根据文件名进行追踪，并保持重试，即该文件被删除或改名后，如果再次创建相同的文件名，会继续追踪。
2.  tail -f。等同于 -follow=descriptor，根据文件描述符进行追踪，当文件改名或被删除，追踪停止。在实际工作中我们的日志数据一般都会通过 log4j 记录，log4j 产生的日志文件名称是固定的，每天定时给文件重命名。假设默认 log4j 会向 access.log 文件中写日志，每当凌晨0点的时候，log4j 都会对文件进行重命名，在 access 后面添加昨天的日期，然后再创建新的 access.log 记录当天的新增日志数据。这个时候如果想要一直监控 access.log 文件中的新增日志数据的话，就需要使用 tail -F。

NetCat TCP/UDP Source：采集指定端口(tcp、udp)的数据，可以读取流经端口的每一行数据。  
Spooling Directory Source：采集文件夹里新增的文件。  
Kafka Source：从 Kafka 消息队列中采集数据。

**Exec Source 和 Kafka Source 在实际工作中是最常见的，可以满足大部分的数据采集需求。**

(5) Channel：数据通道，是中转事件的一个临时存储，保存由 Source 组件传递过来的事件，直到被一个 Sink 消耗。Channel 在 Source 和 Sink 之间连接，起到桥梁的作用。Flume 对于 Channel，则提供了 Memory Channel、JDBC Chanel、File Channel 等技术。

1.  Memory Channel：可以实现高速的吞吐(效率高、不涉及磁盘IO)，但是无法保证数据的完整性，也会存在内存不够用的情况，Memory Channel 在官方文档上已经建议使用 FileChannel 来替换。**非常适合需要高吞吐量但一旦失败会丢失数据的场景。**
2.  JDBC Channel：事件被持久存储在可靠的数据库中，如果可恢复性非常的重要可以使用这种方式。
3.  File Channel：保证数据的完整性与一致性。建议 File Channel 设置的目录和程序日志文件保存的目录设成不同的磁盘，以便提高效率。一般性能会比较低(但是并没有我们想象中的那么慢)，但即使程序出错数据也不会丢失。
4.  Spillable Memory Channel：使用内存和文件作为数据存储，即先把数据存到内存中，如果内存中数据达到阈值再 flush 到文件中。解决了内存不够用的问题，但还是存在数据丢失的风险。

(6) Sink：数据汇聚，代表外部数据存放位置。从 Channel 中读取并移除事件，将事件发送到指定的外部目标(如另一个 Source，也可能是 Hbase、HDFS、Kafka)。Sink 在设置存储数据时，可以向文件系统、数据库、 Hadoop 中储数据。在日志数据较少时，可以将数据存储在文件系中，并且设定一定的时间间隔保存数据。在日志数据较多时，可以将相应的日志数据存储到 Hadoop 中，便于日后进行相应的数据分析。Flume 也提供了各种 Sink 的实现，包括 HDFS Sink、Logger Sink、Avro Sink、File Roll Sink、Null Sink、HBase Sink等。**注意：一个 Sink 仅能接收一个 Channel 的事件。**

Logger Sink：将数据作为日志处理，可以选择打印到控制台或者写到文件中，这个主要在测试的时候使用。  
HDFS Sink：将数据传输到 HDFS 中，这个是比较常见的，主要针对离线计算的场景。  
Kafka Sink：将数据发送到 kafka 消息队列中，这个也是比较常见的，主要针对实时计算场景，数据不落盘，实时传输，最后使用实时计算框架直接处理。

(7) Flume 的结构流程为： Source 接收数据，当 Source 捕获数据后会进行特定的格式化，产生事件，然后 Source 会把事件推入(单个或多个)到缓冲区(Channel)中。Channel 将保存事件直到 Sink 处理完该事件，Sink 将事件发送到指定的外部目标。如果把 Flume 比作情报人员，情报人员的工作为：搜集信息、传递信息、记忆信息，即存储信息。那么 Flume 中，Source 实现了搜集信息，Channel 实现了传递信息，Sink 实现了存储信息。

**Flume 的其他结构：**

Flume 分布式系统中最核心的角色是 Agent，Flume 系统就是由一个个 Agent 所连接起来形成。根据 Flume 中 Agent 的个数，Flume 的体系结构可以有多种，下面分别进行介绍。

1.  单一 Agent 结构。单一 Agent 结构即只有一个 Agent 的结构，是 Flume 的最基本的结构在前面已经介绍过，这里不再进行赘述。
2.  Agent 的顺序结构。如果有两个 Agent，那么 Flume 的结构如下图所示：  
    ![在这里插入图片描述][20210702010323128.png]
3.  Agent 的合并结构。如果有多个 Agent 需要连接的话，一般使用合并 Agent 的结构，即多个 Agent 的数据汇聚到同一个 Agent，这种结构如下图所示：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 1]  
    这种连接方式应用的场景比较多，例如要收集 Web 网站的用户行为日志，Web 网站为了可用性所使用的负载集群模式，每个节点都会产生用户行为日志，可以为每个节点都配置一个 Agent 来单独收集日志数据，然后多个 Agent 将数据最终汇聚到一个用来存储数据存储系统，如 HDFS 上。
4.  Agent 的多路复用结构。Flume 中还支持 Agent 的多路复用结构，当syslog、java、nginx、tomcat 等混合在一起的日志流入一个 Agent 后，可以在 Agent 中将混杂的日志分开，然后给每种日志建立一个自己的传输通道，如下图所示：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 2]

**总结： 一个完整的 Flume 是由以下六个模块组成的。**

1.  外部输入数据。
2.  Event(事件)。
3.  Source(数据收集组件)。
4.  Channel(数据通道)。
5.  Sink(数据汇聚)。
6.  输出数据。

Flume 结构与水槽很类似，如下图所示：  
![在这里插入图片描述][20210702011309461.png]  
分析一下水槽中各部分的场景对应在 Flume 哪一部分结构。

1.  从水厂供到家中的水：外部数据。
2.  水龙头：Source。
3.  水槽中的盛水盆：Channel。
4.  暂存在盛水盆中的水：Event。
5.  盛水盆下方的管道：Sink。
6.  从管道流出的水：输出数据。

**Flume 的特点与优势：**

Flume 是一个分布式、可靠和高可用的海量日志采集、聚合和传输的系统，支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume 提供对数据进行简单处理，并写到各种数据接受方(比如文本、HDFS、Hbase等)的能力 。Flume具有以下的特点：

1.  可靠性。当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume 提供了三种级别的可靠性保障，从强到弱依次分别为：End to End：收到数据 Agent 首先将 Event 写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送。Store on failure：这也是 Scribe(Facebook的开源日志收集系统)采用的策略，当数据接收方宕机时，将数据写到本地，待恢复后，继续发送。Besteffort：数据发送到接收方后，不会进行确认。
2.  可扩展性。Flume 中的组件数目可扩展。
3.  可管理性。可以动态地增加和删除组件。
4.  可恢复性。通道可以用内存或文件的方式实现。内存更快，但是不可恢复；文件比较慢但提供了可恢复性。

Flume 与其他日志收集系统相比，具有以下优势：

1.  Flume 可以将应用产生的数据存储到任何集中存储器中，比如 HDFS，HBase 等。
2.  当收集数据的速度超过将写入数据的时候，也就是当收集信息遇到峰值时，这时候收集的信息非常大，甚至超过了系统的写入数据能力，Flume 会在数据生产者和数据收容器间做出调整，保证其能够在两者之间提供平稳的数据。
3.  Flume 的管道是基于事务，保证了数据在传送和接收时的一致性。

**安装并配置 Flume：**

重新克隆一台 Linux 机器，主机名设置为 bigdata04，ip 设置为192.168.182.103，关闭防火墙，安装 jdk 并配置环境变量，因为 Flume 是 java 开发，所以需要依赖 jdk 环境。这些基本配置不会的话，可以查看 [2021年 全网最细大数据学习笔记（二）：Hadoop 伪分布式安装][2021_ _Hadoop] 一文。

点击 [此处][Link 1] 进行下载。Flume 的安装包有两种，一种 xxx-src.tgz，为 Flume 的源代码，需要自行编译安装，较为灵活；另一种 xxx-bin.tar.gz，为已经编译好的，可以直接使用的安装包。这里我们直接选择第二种，已经被编译过的安装包进行下载安装，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 4]  
确定了要下载的 Flume 版本之后，单击 `apache-flume-1.9.0-bin.tar.gz` 链接，进入到下一页面，如下图所示，单击 `https://mirrors.bfsu.edu.cn/apache/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz` 进行下载。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 5]  
安装包下载好以后上传到 linux 机器的 /data/soft 目录下，并且解压，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 6]  
tar -zxvf apache-flume-1.9.0-bin.tar.gz。修改 flume 的 env 环境变量配置文件。在 flume 的 conf 目录下，修改 flume-env.sh.template 的名字，去掉后缀 template，命令：

[root@bigdata04 soft]# ls
    apache-flume-1.9.0-bin  apache-flume-1.9.0-bin.tar.gz  jdk1.8  jdk-8u202-linux-x64.tar.gz
    [root@bigdata04 soft]# cd apache-flume-1.9.0-bin
    [root@bigdata04 apache-flume-1.9.0-bin]# ls
    bin        conf      doap_Flume.rdf  lib      NOTICE     RELEASE-NOTES
    CHANGELOG  DEVNOTES  docs            LICENSE  README.md  tools
    [root@bigdata04 apache-flume-1.9.0-bin]# cd conf/
    [root@bigdata04 conf]# mv flume-env.sh.template flume-env.sh

使用命令 vi flume-env.sh(注意先要进入到 conf 目录) 打开 flume-env.sh 文件中添加 JAVA\_HOME，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 7]  
创建 Flume 的软链接的命令：ln -s apache-flume-1.9.0-bin/ flume。修改 .bashrc 文件，vi ~/.bashrc。在文件末尾添加如下代码：

export FLUME_HOME=/data/soft/apache-flume-1.9.0-bin
    export PATH=$FLUME_HOME/bin:$PATH

source ~/.bashrc。然后使用 flume-ng version 命令查看 Flume 的版本，如果出现了正确的版本号，则说明 Flume 安装成功，命令如下：

[root@bigdata04 conf]# flume-ng version
    Flume 1.9.0
    Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
    Revision: d4fcab4f501d41597bc616921329a4339f73585e
    Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018
    From source with checksum 35db629a3bda49d23e9b3690c80737f9

如果系统里安装了 HBase，就会出现错误：找不到或无法加载主类。解决办法为：进入到 HBase 安装目录的 conf 目录下，打开 hbase-env.sh 文件，将 `export HBASE_CLASSPATH....` 的代码前添加一个\#，即将此行配置注释掉。注意：如果没有安装 HBase，则不会出现此错误。

# 三、极速上手 Flume 使用 #

## 3.1 案例：Flume 的 Hello World! ##

先来看一个入门级别的 Hello World 案例。启动 Flume 任务其实就是启动一个 Agent，Agent 是由 source、channel、sink 组成的，这些组件在使用的时候只需要写几行配置就可以了。那 source、channel、sink 该如何配置呢？打开官网，找到左边的 Documentation，查看文档信息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 8]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 9]  
进入 Flume User Guide：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 10]  
找到下方的一个入门示例，A simple example，Here, we give an example configuration file, describing a single-node Flume deployment. This configuration lets a user generate events and subsequently logs them to the console.

# example.conf: A single-node Flume configuration  
    
    # Name the components on this agent  定义三大组件的名称
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source 配置source组件
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    # Describe the sink 配置sink组件
    a1.sinks.k1.type = logger
    
    # Use a channel which buffers events in memory 配置和描述Channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel 将Source和Sink绑定到Channel上面
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

这个例子中首先定义了 source 的名字、sink 的名字还有 channel 的名字，下面配置 source 的相关参数，下面配置了 sink 的相关参数，接着配置了 channel 的相关参数，最后把这三个组件连接到了一起，就是告诉 source 需要向哪个 channel 写入数据，告诉 sink 需要从哪个 channel 读取数据，这样 source、channel、sink 这三个组件就联通了。总结下来，配置 Flume agent 的主要流程是这样的：

1.  给每个组件起名字
2.  配置每个组件的相关参数
3.  把它们联通起来

注意了，在 Agent 中配置的三大组件为什么要这样写呢？如果是第一次使用我也不会写啊。三大组件的配置在文档中是有详细说明的，来看一下，在 Flume Sources 下面显示的都是已经内置支持的 Source 组件：  
![20210703135720198.png][]  
刚才看的案例中使用的是 source 类型是 netcat，其实就是 NetCat TCP Source，看一下详细内容：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 11]  
这里面的粗体字体是必选的参数。第一个参数是为了指定 source 需要向哪个 channel 写数据，这个其实是通用的参数，主要看下面这三个，type、bind、port。type：类型需要指定为 natcat，bind：指定当前机器的 ip，使用 hostname 也可以，port：指定当前机器中一个没有被使用的端口。指定 bind 和 port 表示开启监听模式，监听指定 ip 和端口中的数据，其实就是开启了一个 socket 的服务端，等待客户端连接进来写入数据。在这里给 agent 起名为 a1，所以 netcat 类型的配置如下，这里面还指定了 source、channel 的名字，并且把 source 和 channel 连接到一起了，抛开这几个配置之外就剩下了三行配置，就是刚才我们分析的那三个必填参数。

a1.sources = r1
    a1.channels = c1
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 6666
    a1.sources.r1.channels = c1

注意了，bind 参数后面指定的 ip 是四个 0，这个当前机器的通用 ip，因为一台机器可以有多个 ip，例如：内网ip、外网ip，如果通过 bind 参数指定某一个 ip 的话，表示就只监听通过这个 ip 发送过来的数据了，这样会有局限性，所以可以指定 0.0.0.0。下面几个参数都是可选配置，默认可以不配置。接着是 channel，案例中 channel 使用的是 memory。  
![20210703140953286.png][]  
查看 memory channel：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 12]  
这里面只有 type 是必填项，其他都是可选的。最后看一下 sink，在案例中 sink 使用的是 logger，对应的就是 Logger Sink：  
![在这里插入图片描述][20210703141345902.png]  
logger sink 中默认也只需要指定 type 即可，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 13]  
后期我们如果想要使用其他的内置组件，直接到官网文档这里查找即可，这里面的配置有很多，没有必要去记，肯定记不住，只要知道到哪里去找就可以，官网是可以随便使用的，所以建议大家到官网找到配置之后直接拷贝，要不然自己手写很容易出错。配置文件分析完了，可以把这些配置放到一个配置文件中，起名叫 example.conf，把这个配置文件放到 conf/ 目录下。

[root@bigdata04 ~]# cd /data/soft/apache-flume-1.9.0-bin/conf/
    [root@bigdata04 conf]# vi example.conf 
    mple.conf: A single-node Flume configuration
    
    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    # Describe the sink
    a1.sinks.k1.type = logger
    
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    ~                                                                                               
    "example.conf" 23L, 540C
    mple.conf: A single-node Flume configuration
    
    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    # Describe the sink
    a1.sinks.k1.type = logger
    
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

注意了，这个配置文件中的 a1 表示是 agent 的名称，还有就是 port 指定的端口必须是未被使用的，可以先查询一下当前机器使用了哪些端口，端口的可用范围是1-65535，如果懒得去查的话，就尽量使用偏大一些的端口，这样被占用的概率就非常低了。Agent 配置好了以后就可以启动了，下面来看一下启动 Agent 的命令：

[root@bigdata04 conf]# cd ..
    [root@bigdata04 apache-flume-1.9.0-bin]#  bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf -Dflume.root.logger=INFO,console

这里面使用 flume-ng 命令，解释如下：

后面指定 agent，表示启动一个Flume的agent代理
    --name：指定 agent 的名字
    --conf：指定 flume 配置文件的根目录
    --conf-file：指定 Agent 对应的配置文件(包含source、channel、sink配置的文件)
    -D：动态添加一些参数，在这里是指定了flume的日志输出级别和输出位置，INFO表示日志级别，console表示是控制台的意思，
    也就是说默认会把日志数据打印到控制台上，方便查看，一般在学习测试阶段会指定这个参数

注意了，其实 agent 的启动命令还可以这样写：

bin/flume-ng agent -n $agent_name -c conf -f conf/flume-conf.properties.template
    这里面的-n属于简写，完整的写法就是 --name
    -c 完整写法的 --conf
    -f 完整写法是 --conf-file

所以以后看到这两种写法要知道它们都是正确的写法。启动 Agent，在这里我们使用完整的写法，看起来清晰一些：

[root@bigdata04 apache-flume-1.9.0-bin]#  bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf -Dflume.root.logger=INFO,console

启动之后会看到如下信息，表示启动成功，启动成功之后，这个窗口会被一直占用，因为 Agent 服务一直在运行，现在属于—个前台进程。

2021-07-03 14:28:14,708 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.source.NetcatSource.start(NetcatSource.java:166)] Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]

如果看到提示的有 ERROR 级别的日志信息，就需要具体问题具体分析了，一般都是配置文件配置错误了。接下来我们需要连接到 source 中通过 netcat 开启的 socket 服务端，克隆一个 bigdata04 的会话，因为前面启动 Agent 之后，窗口就被占用了，使用 **telnet命令** 可以连接到指定 socket 服务，telnet 后面的主机名和端口是根据 **example.conf 配置文件** 中配置的。**注意：如果提示找不到 telnet 命令，则需要使用 yum 在线安装。**

[root@bigdata04 ~]# telnet localhost 44444
    Trying ::1...
    telnet: connect to address ::1: Connection refused
    Trying 127.0.0.1...
    Connected to localhost.
    Escape character is '^]'.
    Hello World!
    OK

回到 Agent 所在的窗口，可以看到下面多了一行日志，就是我们在 telnet 中输入的内容：  
![在这里插入图片描述][20210703143159472.png]  
按 ctrl+c 断开 telnet 连接，重新使用 telnet 连接，此时不使用 localhost，使用本机的内网ip可以吗? 192.168.61.103：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 14]  
所以此时 Agent 中 source 的配置在使用的时候就受限制了，在开启 telnet 客户端的时候就只能在本地开启了，无法在其他机器上使用，因为 source 中绑定的 ip 是 localhost。如果想要支持一个网络内其它机器上也可以使用 telnet 链接的话就需要修改 bind 参数指定的值了，最直接的就是指定192.168.61.103 这个内网 ip，其实还有一种更加通用的方式是指定 0.0.0.0，此时表示会监听每一个可用的 ip 地址，所以在绑定 ip 端口时，ip 通常都会使用 0.0.0.0，那在这里我们把 example.conf 中的 localhost 改为0.0.0.0，按 ctrl+c 停止刚才启动的 agent。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 15]  
启动 Agent，在另一个会话窗口中使用 telnet 连接，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 16]  
此时可以在其他机器上使用 telnet 连接也可以，在 bigdata01 机器上，

[root@bigdata01 ~]# yum install -y telnet
    [root@bigdata01 ~]# telnet 192.168.61.103 44444
    Trying 192.168.61.103...
    Connected to 192.168.61.103.
    Escape character is '^]'.
    I am bigdata01
    OK

这就是 Flume 中的 Hello World 了。但是注意了，此时 Flume 中 Agent 服务是在前台运行，这个服务实际工作中需要一直运行，所以需要放到后台运行。Flume 自身没有提供直接把进程放到后台执行的参数，需要使用 **nohup和&** 了。此时就不需要指定 **-Dflume.root.logger=lNFO,console** 参数了，默认情况下 flume 的日志会记录到日志文件中。停掉之前的 Agent，重新执行。

[root@bigdata04 apache-flume-1.9.0-bin]# nohup bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf &

启动之后，通过 jps 命令可以查看到一个 application 进程，这个就是启动的 Agent：  
![在这里插入图片描述][20210703144736119.png]  
这样看起来不清晰，如果后期启动了多个 Agent，都分不出来哪个是哪个了可以在 jps 后面加上参数 -m，这样可以看到启动时指定的一些参数信息：

[root@bigdata04 apache-flume-1.9.0-bin]# jps -m
    2346 Application --name a1 --conf-file conf/example.conf
    2395 Jps -m

或者使用 ps 命令也可以：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 17]  
这个 Agent 中的 sink 组件把数据以日志的方式写出去了，所以这个数据默认就会进入到 flume 的日志文件中，那我们来看一下 flume 的日志文件，在 flume 的 logs 目录中有一个 flume.log：

[root@bigdata04 apache-flume-1.9.0-bin]# cd logs/
    [root@bigdata04 logs]# tail -2 flume.log 
    03 Jul 2021 14:45:54,561 INFO  [lifecycleSupervisor-1-0] (org.apache.flume.source.NetcatSource.start:155)  - Source starting
    03 Jul 2021 14:45:54,563 INFO  [lifecycleSupervisor-1-0] (org.apache.flume.source.NetcatSource.start:166)  - Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/0:0:0:0:0:0:0:0:44444]

再使用 telnet 向里面输入一些数据，查看一下日志，可以看到数据，是没有问题的，如果后期发现 Agent 服务突然不干活了，或者对应的进程没有了，就需要到这里来查看日志了。此时想要停止这个 Agent 的话就需要使用 kill 命令了，如下：  
![在这里插入图片描述][20210703145308190.png]

## 3.2 案例：采集文件内容上传至 HDFS ##

**需求：** 某服务器的某特定目录下，会不断产生新的文件，每当有新文件出现，就需要把文件中的内容采集到 HDFS 中去。根据需求，首先定义以下 3 大要素。

1.  采集源，即 Source——监控文件目录：Spooldir。其作用为，监控一个目录，只要此目录中出现新文件，就会采集文件中的内容。采集完成，文件会被 agent 自动添加一个后缀，即 **.COMPLETED**，缺点是不支持老文件新增数据的收集。注意：所监控的目录中不允许重复出现相同文件名的文件。
2.  下沉目标，即 Sink——HDFS 文件系统： HDFS Sink。
3.  Source 和 Sink 之间的传递通道——Channel：可用 File Channel 也可以用内存 Channel。

**实现：** 首先是基于目录的 source，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 18]  
channels 和 type 肯定是必填的，还有一个是 spoolDir，就是指定一个监控的目录，看它下面的案例，里面还多指定了一个 fileHeader，这个暂时不用，配置 source：

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.c1.type = spooldir
    a1.sources.c1.spoolDir = /data/log/studentDir

接下来是 channel 了，channel 在这里使用基于文件的，可以保证数据的安全性。如果针对采集的数据，丢个两条对整体结果影响不大，只要求采集效率，那么这个时候完全可以使用基于内存的 channel，这里使用基于文件的 channel，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 19]  
根据这里的例子可知，要配置 checkpointDir 和 dataDir，因为这两个目录默认会在用户家录下生成，建议修改到其他地方，checkpointDir 是存放检查点的目录，data 是存放数据的目录。

a1.channels.c1.type = file
    a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
    a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data

最后是 sink，因为要向 hdfs 中输出数据，所以可以使用 hdfs sink。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 20]  
参数解析如下表所示：

<table> 
 <thead> 
  <tr> 
   <th align="left">参数</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">type</td> 
   <td align="left">Sinks的类型，这里为HDFS，指将数据采集到HDFS中</td> 
  </tr> 
  <tr> 
   <td align="left">path</td> 
   <td align="left">写入HDFS文件的目录，需要包含文件系统标识，比如：hdfs://namenode/flume/webdata/，使用Flume提供的日期及%{host}表达式</td> 
  </tr> 
  <tr> 
   <td align="left">filePrefix</td> 
   <td align="left">放置文件的前缀，为access_log。还有fileSuffix参数是设置放置文件的后缀，如.lzo，.log等</td> 
  </tr> 
  <tr> 
   <td align="left">maxOpenFiles</td> 
   <td align="left">允许打开的HDFS的文件数，当打开的文件数达到该值时，最早打开的文件将会被关闭，默认为5000</td> 
  </tr> 
  <tr> 
   <td align="left">batchSize</td> 
   <td align="left">每个批次刷新到HDFS上的events数量</td> 
  </tr> 
  <tr> 
   <td align="left">fileType</td> 
   <td align="left">定义文件格式，包括SequenceFile，DataStream，CompressedStream当使用DataStream时候，文件不会被压缩，不需要设置hdfs.codeC当使用CompressedStream时候，必须设置一个正确的hdfs.codeC值注意：写入的文件和读取的文件类型保持一致</td> 
  </tr> 
  <tr> 
   <td align="left">writeFormat</td> 
   <td align="left">写sequence文件的格式。包含：Text, Writable（默认）</td> 
  </tr> 
  <tr> 
   <td align="left">rollSize</td> 
   <td align="left">当临时文件达到该大小（单位：bytes）时，滚动成目标文件默认值为1024如果设置成0，则表示不根据临时文件大小来滚动文件</td> 
  </tr> 
  <tr> 
   <td align="left">rollCount</td> 
   <td align="left">当events数据达到该数量时候，将临时文件滚动成目标文件默认值为10 如果设置成0，则表示不根据events数据来滚动文件</td> 
  </tr> 
  <tr> 
   <td align="left">rollInterval</td> 
   <td align="left">HDFS Sink间隔多长将临时文件滚动成最终目标文件，单位：秒默认值为30如果设置成0，则表示不根据时间来滚动文件注意：滚动（roll）指的是，hdfs sink 将临时文件重命名成最终目标文件，并新打开一个临时文件来写入数据</td> 
  </tr> 
  <tr> 
   <td align="left">round</td> 
   <td align="left">是否启用时间上的“舍弃”，这里的“舍弃”，类似于“四舍五入”，后面再介绍。如果启用，则会影响除了%t的其他所有时间表达式默认值为false</td> 
  </tr> 
  <tr> 
   <td align="left">roundValue</td> 
   <td align="left">时间上进行“舍弃”的值默认值为1</td> 
  </tr> 
  <tr> 
   <td align="left">roundUnit</td> 
   <td align="left">时间上进行“舍弃”的单位，包含：second,minute,hour默认值为seconds在这里因为设置的是舍弃10分钟内的时间，因此该目录每10分钟新生成一个</td> 
  </tr> 
  <tr> 
   <td align="left">useLocalTimeStamp</td> 
   <td align="left">是否使用当地时间默认值为flase</td> 
  </tr> 
 </tbody> 
</table>

配置 Sinks，代码如下：

# Describe the sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/flume/studentDir
    a1.sinks.k1.hdfs.filePrefix = stu-
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0‬

最后把组件连接到一起：

# Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

把 Agent 的配置保存到 flume 的 conf 目录下的 file-to-hdfs.conf 文件中：

# example.conf: A single-node Flume configuration
    
    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = spooldir
    a1.sources.r1.spoolDir = /data/log/studentDir
    
    # Describe the sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/flume/studentDir
    a1.sinks.k1.hdfs.filePrefix = stu-
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0‬
    
    
    # Use a channel which buffers events in memory
    a1.channels.c1.type = file
    a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
    a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data
    
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

下面就可以启动 Agent 了，在启动 Agent 之前，先初始化一下测试数据，创建 /data/log/studentDir 目录，然后在里面添加一个文件，class1.dat，class1.dat 中存储的是学生信息，学生姓名、年龄、性别，如下：

[root@bigdata04 log]# mkdir -p /data/log/studentDir
    [root@bigdata04 log]# cd /data/log/studentDir/
    [root@bigdata04 studentDir]# vi class1.dat
    jack 18 male
    jessic 20 female
    tom 17 male
    paul 25 male
    jerry 30 female
    [root@bigdata04 studentDir]# more class1.dat         
    jack 18 male
    jessic 20 female
    tom 17 male
    paul 25 male
    jerry 30 female

**注意：在开启 Agent 之前，必须首先开启 Hadoop。并且把集群中修改好配置的 hadoop 目录远程拷贝到 bigdata04 上。拷完成之后到 bigdata04 节点上验证一下，如下：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 21]  
**注意：还需要修改环境变量，配置 HADOOP\_ HOME，否则启动 Agent 的时候还是会报错。**

启动 Agent，此时可以看到 Agent 正常启动，还可以看到类似这样的日志：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 22]  
到 hdfs 上验证结果，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 23]  
此时发现文件已经生成了，只不过默认情况下现在的文件是 `.tmp` 结尾的，表示它在被使用，因为 Flume 只要采集到数据就会向里面写，这个后缀默认是由 `hdfs.inUsesuffix` 参数来控制的。文件名上还拼接了一个当前时间戳，这个是默认文件名的格式，当达到文件切割时机的时候会给文件改名字，去掉 `.tmp`，这个文件现在也是可以查看的，里面的内容其实就是 class1.dat 文件中的内容。如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 24]  
所以此时 Flume 就会监控 linux 中的 /data/log/studentDir 目录，当发现里面有新文件的时候就会把数据采集过来。那 Flume 怎么知道哪些文件是新文件呢?它会不会重复读取同一个文件的数据呢?不会的，我们到 /data/log/studentDir 目录看一 下你就知道了，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 25]  
我们发现此时这个文件已经被加了一个后缀 `.COMPLETED` ，表示这个文件已经被读取过了，所以 Flume 在读取的时候会忽略后缀为 `.COMPLETED` 的文件。接着我们再看一下 channel 中的数据， 因为数据是存在本地磁盘文件中的，所以是可以去看一下的， 进入 dataDir 指定的目录：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 26]  
发现里面有类似 `log-N` 的文件，这个文件中存储的其实就是读取到的内容，不过在这无法直接查看。现在我们想看一下 Flume 最终生成的文件是什么样子的，停止 Agent 就可以看到了(暴力操作)，当 Agent 停止的时候就会去掉 `.tmp` 标志了。  
![在这里插入图片描述][20210703185910657.png]  
当再重启 Agent 之后，不会再给加上 `.tmp` 了，每次停止之前都会把所有的文件解除占用状态，下次启动的时候如果有新数据，则会产生新的文件，这其实就模拟了一下自动切文件之后的效果。但是这个文件看起来比较别扭，连个后缀都没有，没有后缀倒不影响使用，就是看起来不好看，大家可以自己去查看官网文档资料，修改 Agent 配置，添加测试数据，验证效果。

## 3.3 案例：采集网站日志上传至 HDFS ##

**需求：** 将 bigdata02 和 bigdata03 两台机器实时产生的日志数据汇总到机器 bigdata04 中，通过机器 bigdata04 将数据统一上传至 HDFS 的指定目录中。看下面这个图，来详细分析一下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 27]  
根据刚才的需求分析可知，我们一共需要三台机器，这里使用 bigdata02 和 bigdata03 采集当前机器产生的实时日志数据，统一汇总到 bigdata04 机器上。其中 bigdata02 和 bigdata03 中的 source 使用基于 file 的 source，ExecSource，因为要实时读取文件中的新增数据，channel 在这里我们使用基于内存的 channel，因为这里是采集网站的访问日志，就算丢一两条数据对整体结果影响也不大，我们只希望采集到的数据可以快读进入 hdfs 中，所以就选择了基于内存的 channel。由于 bigdata02 和 bigdata03 的数据需要快速发送到 bigdata04 中，为了快速发送我们可以通过网络直接传输，sink 建议使用 avrosink，avro 是一种数据序列化系统，经过它列化的数据传输起来效率更高，并且它对应的还有一个 avrosource，avrosink 的数据可以直接发送给 avrosource，所以他们可以无缝衔接。这样 bigdata04 的 source 就确定了使用 avrosource、channel 还是基于内存的 channel，sink 就使用 hdfs sink，因为是要向 hdfs 中写数据的。这里面的组件，只有 execsource、 avrosource、 avrosink 在之前还没有使用过， 其他的组件都使用过了。最终需要在每台机器上启动一个 agent，启动的时候需要注意先后顺序，先启动 bigdata04 上面的，再启动 bigdata02 和 bigdata03 上面的。

**实现：**

(1) 将 bigdata04 中配置好的 apache-flume-1.9.0-bin 安装目录拷贝到 bigdata02 以及 bigdata03 机器上。

[root@bigdata04 soft]# scp -rq apache-flume-1.9.0-bin bigdata02:/data/soft/
    [root@bigdata04 soft]# scp -rq apache-flume-1.9.0-bin bigdata03:/data/soft/

(2) 配置 bigdata02 上的 Agent，创建文件 file-to-avro-101.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /data/log/access.log
    
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = 192.168.61.103
    a1.sinks.k1.port = 45454
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

这里面的配置没有特殊配置，直接参考官网文档就可以搞定。

(3) 配置 bigdata03 上的 Agent，创建文件 file-to-avro-102.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /data/log/access.log
    
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = 192.168.61.103
    a1.sinks.k1.port = 45454
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

(4) 配置 bigdata04 上的 Agent，创建文件 avro-to-hdfs.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 45454
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/access/%Y%m%d
    a1.sinks.k1.hdfs.filePrefix = access
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

三台机器中的 Flume Agent 都配置好了，在开始启动之前需要先在 bigdata02 和 bigdata03 中生成测试数据，为了模拟真实情况，在这里我们就开发一个脚本， 定时向文件中写数据：

#!/bin/bash
    # 循环向文件中生成数据
    while [ "1" = "1" ]
    do
    	# 获取当前时间戳
    	curr_time=`date +%s`
    	# 获取当前主机名
    	name=`hostname`
    	echo ${name}_${curr_time} >> /data/log/access.log
    	# 暂停1秒
    	sleep 1
    done

(5) 在 bigdata02 和 bigdata03 中使用这个脚本生成数据，首先在 bigdata02 上创建 /data/log 目录，然后创建 `generateAccessLog.sh` 脚本，如下：

[root@bigdata02 ~]# mkdir -p /data/log
    [root@bigdata02 ~]# cd /data/log/
    [root@bigdata02 log]# vi generateAccessLog.sh

同样在 bigdata03 上创建 /data/log 目录，然后创建 `generateAccessLog.sh` 脚本。接下来开始启动相关的服务进程，首先启动 bigdata04 上的 agent 服务，

bin/flume-ng agent --name a1 --conf conf --conf-file conf/avro-to-hdfs.conf -Dflume.root.logger=INFO,console

接下来启动 bigdata02 上的 agent 服务和 shell 脚本：

bin/flume-ng agent --name a1 --conf conf --conf-file conf/file-to-avro-101.conf -Dflume.root.logger=INFO,console
    sh -x generateAccessLog.sh

最后启动 bigdata03 上的 agent 服务和 shell 脚本：

bin/flume-ng agent --name a1 --conf conf --conf-file conf/file-to-avro-102.conf -Dflume.root.logger=INFO,console
    sh -x generateAccessLog.sh

验证结果，查看 hdfs 上的结果数据，在 bigdata01 上查看：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 28]  
注意：启动之后稍等一会就可以看到数据了，我们观察数据的变化，会发现 hdfs 中数据增长的不是很快，它会每隔一段时间添加一批数据，实时性好像没那么高？这是因为 avrosink 中有一个配置 batch-size，它的默认值是100，也就是每次发送100条数据，如果数据不够100条，则不发送。具体这个值设置多少合适，要看你 source 数据源大致每秒产生多少数据，以及你希望的延迟要达到什么程度，如果这个值设置太小的话，会造成 sink 频繁向外面写数据，这样也会影响性能。最终，依次停止 bigdata02、bigdata03 中的服务，最后停止 bigdata04 中的服务。

# 四、Flume 高级组件 #

**高级组件：**

1.  Source Interceptors：Source 可以指定一个或者多个拦截器按先后顺序依次对采集到的数据进行处理。
2.  Channel Selectors：Source 发往多个 Channel 的策略设置，如果 source 后面接了多个 channel，到底是给所有的 channel 都发，还是根据规则发送到不同 channel，这些是由 Channel Selectors 来控制的。
3.  Sink Processors：Sink 发送数据的策略设置，一个 channel 后面可以接多个 sink，channel 中的数据是被哪个 sink 获取，这个是由 Sink Processors 控制的。

## 4.1 Source Interceptors ##

系统中已经内置提供了很多 Source Interceptors，常见的 Source Interceptors 类型：Timestamp Interceptor、Host Interceptor、Search and Replace Interceptor 、Static Interceptor、Regex Extractor Interceptor等。

1.  Timestamp Interceptor：向 event 中的 header 里面添加 timestamp 时间戳信息。
2.  Host Interceptor：向 event 中的 header 里面添加 host 属性，host 的值为当前机器的主机名或者ip。
3.  Search and Replace Interceptor：根据指定的规则查询 Event 中 body 里面的数据，然后进行替换，这个拦截器会修改 event 中 body 的值，也就是会修改原始采集到的数据内容。
4.  Static Interceptor：向 event 中的 header 里面添加固定的 key 和 value。
5.  Regex Extractor Interceptor：根据指定的规则从 Event 中的 body 里面抽取数据，生成 key 和 value，再把 key 和 value 添加到 header 中。

根据刚才的分析，总结一下：Timestamp Interceptor、Host Interceptor、Static Interceptor、Regex Extractor Interceptor 是向 event 中的 header 里面添加 key-value 类型的数据，方便后面的 channel 和 sink 组件使用，对采集到的原始数据内容没有任何影响。Search and Replace Interceptor 是会根据规则修改 event 中 body 里面的原始数据内容，对 header 没有任何影响，使用这个拦截器需要特别小心，因为它会修改原始数据内容。这里面这几个拦截器，其中 Search and Replace Interceptor 和 Regex Extractor Interceptor 在工作中使用的比较多一些。下面呢，来看一个案例：对采集到的数据按天按类型分目录存储，原始数据是这样的，看这个文件，Flume测试数据格式.txt：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 29]  
这份数据中有三种类型的数据，视频信息、用户信息、送礼信息，数据都是 json 格式的，这些数据还有一个共性就是里面都有一个 type 字段，type 字段的值代表数据类型，当直播平台正常运行的时候，会实时产生这些日志数据，我们希望把这些数据采集到 hdfs 上进行存储，并且要按照数据类型进行分目录存储，视频数据放一块、用户数据放一块、送礼数据放一块，针对这个需求配置 agent 的话，source 使用基于文件的 execsource、channle 使用基于文件的 channle，我们希望保证数据的完整性和准确性，sink 使用 hdfssink，但是注意了，hdfssink 中的 path 不能写死，首先是按天就是需要动态获取日期，然后是因为不同类型的数据要存储到不同的目录中，那也就意味着 path 路径中肯定要是有变量，除了日期变量还要有数据类型变量，这里的数据类型的格式都是单词中间有一个下划线，但是要求是目录中的单词不要出现下划线，使用驼峰的命名格式。所以最终在 hdfs 中需要生成的目录大致是这样的：

[root@bigdata01 hadoop-3.2.0]# hdfs dfs -ls -R /moreType/20210807
    drwxr-xr-x   - root supergroup          0 2021-08-07 16:56 /moreType/20210807/giftRecord
    -rw-r--r--   2 root supergroup        138 2021-08-07 16:56 /moreType/20210807/giftRecord/data.1628326592344.log.tmp
    drwxr-xr-x   - root supergroup          0 2021-08-07 16:56 /moreType/20210807/userInfo
    -rw-r--r--   2 root supergroup        413 2021-08-07 16:56 /moreType/20210807/userInfo/data.1628326592323.log.tmp
    drwxr-xr-x   - root supergroup          0 2021-08-07 16:56 /moreType/20210807/videoInfo
    -rw-r--r--   2 root supergroup        297 2021-08-07 16:56 /moreType/20210807/videoInfo/data.1628326591470.log.tmp

这里的日期变量好获取，但是数据类型如何获取呢？注意了，咱们前面分析了，通过 source 的拦截器可以向 event 的 header 中添加 key-value，然后在后面的 channle 或者 sink 中获取 key-value 的值。那我们在这就可以通过 Regex Extractor Interceptor 获取原始数据中的 type 字段的值，获取出来以后存储到 header 中，这样在 sink 阶段就可以获取到了。但是这个时候直接获取到的 type 的值是不满足要求的，需要对 type 的值进行转换，去掉下划线，转化为驼峰形式，所以可以先使用 Search and Replace Interceptor 对原始数据中 type 的值进行转换，然后使用 Regex Extractor Interceptor 指定规则获取 type 字段的值，添加到 header 中。所以整体的流程是这样的：

Exec Source -> Search and Replace Interceptor->Regex Extractor Interceptor->File Channel->HDFS Sink

那下面我们来配置 Agent，在 bigdata04 机器上创建 file-to-hdfs-moreType.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /data/log/moreType.log
    
    # 配置拦截器 [多个拦截器按照顺序依次执行]
    a1.sources.r1.interceptors = i1 i2 i3 i4
    a1.sources.r1.interceptors.i1.type = search_replace
    a1.sources.r1.interceptors.i1.searchPattern = "type":"video_info"
    a1.sources.r1.interceptors.i1.replaceString = "type":"videoInfo"
    
    a1.sources.r1.interceptors.i2.type = search_replace
    a1.sources.r1.interceptors.i2.searchPattern = "type":"user_info"
    a1.sources.r1.interceptors.i2.replaceString = "type":"userInfo"
    
    a1.sources.r1.interceptors.i3.type = search_replace
    a1.sources.r1.interceptors.i3.searchPattern = "type":"gift_record"
    a1.sources.r1.interceptors.i3.replaceString = "type":"giftRecord"
    
    
    a1.sources.r1.interceptors.i4.type = regex_extractor
    a1.sources.r1.interceptors.i4.regex = "type":"(\\w+)"
    a1.sources.r1.interceptors.i4.serializers = s1
    a1.sources.r1.interceptors.i4.serializers.s1.name = logType
    
    
    # 配置channel组件
    a1.channels.c1.type = file
    a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/moreType/checkpoint
    a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/moreType/data
    
    
    # 配置sink组件
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.182.100:9000/moreType/%Y%m%d/%{logType}
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    
    #增加文件前缀和后缀
    a1.sinks.k1.hdfs.filePrefix = data
    a1.sinks.k1.hdfs.fileSuffix = .log
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

注意：这里面的拦截器，拦截器可以设置一个或者多个，source 采集的每一条数据都会经过所有的拦截器进行处理，多个拦截器按照顺序执行。后面也统一设置了 hdfs 文件的前缀和后缀，先生成测试数据，提前手工把数据添加到 moreType.log 文件中。

cd /data/log/
    vi moreType.log
    {"id":"14943445328940974601","uid":"840717325115457536","lat":"53.530598","lnt":"-2.5620373","hots":0,"title":"0","status":"1","topicId":"0","end_time":"1494344570","watch_num":0,"share_num":"1","replay_url":null,"replay_num":0,"start_time":"1494344544","timestamp":1494344571,"type":"video_info"}
    {"uid":"861848974414839801","nickname":"mick","usign":"","sex":1,"birthday":"","face":"","big_face":"","email":"abc@qq.com","mobile":"","reg_type":"102","last_login_time":"1494344580","reg_time":"1494344580","last_update_time":"1494344580","status":"5","is_verified":"0","verified_info":"","is_seller":"0","level":1,"exp":0,"anchor_level":0,"anchor_exp":0,"os":"android","timestamp":1494344580,"type":"user_info"}
    {"send_id":"834688818270961664","good_id":"223","video_id":"14943443045138661356","gold":"10","timestamp":1494344574,"type":"gift_record"}

启动 Agent：

bin/flume-ng agent --name a1 --conf conf --conf-file conf/file-to-hdfs-moreType.conf -Dflume.root.logger=INFO,console

验证结果，在 bigdata01 节点上查看：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 30]  
看一下 HDFS 中的文件内容，发现 type 字段的值确实被拦截器修改了，这就实现了按天，按类型分目录存储。

## 4.2 Channel Selectors ##

接下来看一下 Channel Selectors，Channel Selectors 类型包括：Replicating Channel Selector 和 Multiplexing Channel Selector，其中 Replicating Channel Selector 是默认的 channel 选择器，它会将 Source 采集过来的 Event 发往所有 Channel，查看官方文档中针对这个默认 channel 选择器的解释。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 31]  
在这个例子的配置中，c3 是可选 channel。对 c3 的写入失败将被忽略。由于 c1 和 c2 未标记为可选，因此未能写入这些 channel 将导致事务失败。针对这个配置，通俗一点来说就是，source 的数据会发往 c1、c2、c3 这三个 channle 中，可以保证 c1、c2 一定能接收到所有数据，但是 c3 就无法保证了。这个 selector.optional 参数是一个可选项，可以不用配置就行。如果是多个 channel 的话，直接在 channels 参数后面指定多个 channel 的名称就可以了，多个 channel 名称中间使用空格隔开，其实你看这个名称是 channels 带有 s，从名字上看就能看出来他支持多个 channel。还有一个 channel 选择器是 Multiplexing Channel Selector，它表示会根据 Event 中 header 里面的值将 Event 发往不同的 Channel：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 32]  
在这个例子的配置中，指定了4个channel：c1、c2、c3、c4，source 采集到的数据具体会发送到哪个 channel 中，会根据 event 中 header 里面的 state 属性的值，这个是通过 selector.header 控制的

1.  如果 state 属性的值是 CZ，则发送给 c1
2.  如果 state 属性的值是 US，则发送给 c2 c3
3.  如果 state 属性的值是其它值，则发送给 c4

这些规则是通过 selector.mapping 和 selector.default 控制的。这样就可以实现根据一定规则把数据分发给不同的 channel 了。

**案例一：多Channel之Replicating Channel Selector**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 33]  
在这个案例中我们使用 Replicating 选择器，将 source 采集到的数据重复发送给两个 channel，最后每个 channel 后面接一个 sink，负责把数据存储到不同存储介质中，方便后期使用。在实际工作中这种需求还是比较常见的，就是希望把一份数据采集过来以后，分别存储到不同的存储介质中，不同存储介质的特点和应用场景是不一样的，典型的就是 hdfs sink 和 kafka sink，通过 hdfs sink 实现离线数据落盘存储，方便后面进行离线数据计算。通过 kafka sink 实现实时数据存储，方便后面进行实时计算(先使用 logger sink 代理)。下面根据图中列出来的 source、channel 和 sink 来配置 agent，在 bigdata04 中 flume 的 conf 目录中创建 tcp-to-replicatingchannel.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1 c2
    a1.sinks = k1 k2
    
    # 配置source组件
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 44444
    
    # 配置channle选择器[默认就是replicating，所以可以省略]
    a1.sources.r1.selector.type = replicating
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.channels.c2.type = memory
    a1.channels.c2.capacity = 1000
    a1.channels.c2.transactionCapacity = 100
    
    
    
    # 配置sink组件
    a1.sinks.k1.type = logger
    
    a1.sinks.k2.type = hdfs
    a1.sinks.k2.hdfs.path = hdfs://192.168.61.100:9000/replicating
    a1.sinks.k2.hdfs.fileType = DataStream
    a1.sinks.k2.hdfs.writeFormat = Text
    a1.sinks.k2.hdfs.rollInterval = 3600
    a1.sinks.k2.hdfs.rollSize = 134217728
    a1.sinks.k2.hdfs.rollCount = 0
    a1.sinks.k2.hdfs.useLocalTimeStamp = true
    a1.sinks.k2.hdfs.filePrefix = data
    a1.sinks.k2.hdfs.fileSuffix = .log
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1 c2
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c2

启动 Agent：\[root@bigdata04 apache-flume-1.9.0-bin\]\# bin/flume-ng agent --name a1 --conf conf --conf-file conf/tcp-to-replicatingchannel.conf -Dflume.root.logger=INFO,console  
生成测试数据，通过 telnet 连接到 socket：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 34]  
查看效果，可以看到 Flume 在控制台输出的日志信息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 35]  
查看 sink2 在 hdfs 中生成的文件中的内容：

[root@bigdata01 hadoop-3.2.0]# hdfs dfs -cat hdfs://192.168.61.100:9000/replicating/data.1628334503559.log.tmp
    hello
    world

**案例二：多Channel之Multiplexing Channel Selector**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 36]  
在这个案例中使用 Multiplexing 选择器，将 source 采集到的数据按照一定规则发送给两个 channle，最终再把不同 channel 中的数据存储到不同介质中。在这里面我们需要用到正则抽取拦截器在 Event 的 header 中生成 key-value，作为 Multiplexing 选择器的规则，假设原始数据格式为：

{"name":"amo","age":20,"city":"sh"}
    {"name":"paul","age":19,"city":"bj"}

下面来配置 Agent，复制 tcp-to-replicatingchannel.conf 的内容，主要增加 source 拦截器和修改 channel 选择器，以及 hdfs sink 中的 path 路径，在 bigdata04上创建新文件 tcp-to-multiplexingchannel.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1 c2
    a1.sinks = k1 k2
    
    # 配置source组件
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 44444
    
    # 配置source拦截器
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = regex_extractor
    a1.sources.r1.interceptors.i1.regex = "city":"(\\w+)"
    a1.sources.r1.interceptors.i1.serializers = s1
    a1.sources.r1.interceptors.i1.serializers.s1.name = city
    
    # 配置channle选择器
    a1.sources.r1.selector.type = multiplexing
    a1.sources.r1.selector.header = city
    a1.sources.r1.selector.mapping.bj = c1
    a1.sources.r1.selector.default = c2
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.channels.c2.type = memory
    a1.channels.c2.capacity = 1000
    a1.channels.c2.transactionCapacity = 100
    
    
    
    # 配置sink组件
    a1.sinks.k1.type = logger
    
    a1.sinks.k2.type = hdfs
    a1.sinks.k2.hdfs.path = hdfs://192.168.61.100:9000/multiplexing
    a1.sinks.k2.hdfs.fileType = DataStream
    a1.sinks.k2.hdfs.writeFormat = Text
    a1.sinks.k2.hdfs.rollInterval = 3600
    a1.sinks.k2.hdfs.rollSize = 134217728
    a1.sinks.k2.hdfs.rollCount = 0
    a1.sinks.k2.hdfs.useLocalTimeStamp = true
    a1.sinks.k2.hdfs.filePrefix = data
    a1.sinks.k2.hdfs.fileSuffix = .log
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1 c2
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c2

启动 Agent：\[root@bigdata04 apache-flume-1.9.0-bin\]\# bin/flume-ng agent --name a1 --conf conf --conf-file conf/tcp-to-multiplexingchannel.conf -Dflume.root.logger=INFO,console  
生成测试数据，通过 telnet 连接到 socket：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 37]  
查看结果，在 Flume 启动命令行中会输出如下日志信息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 38]  
查看 sink2 输出到 hdfs 中的数据：

[root@bigdata01 hadoop-3.2.0]# hdfs dfs -cat hdfs://192.168.61.100:9000/multiplexing/data.1628335976766.log.tmp
    {"name":"ben","age":20,"city":"sh"}

这样就实现了，根据规则把 source 采集到的数据分发到不同 channel 中，最终输出到不同存储介质中，这就是 Multiplexing Channel Selector 的应用了。

## 4.3 Sink Processors ##

接下来看一下 Sink 处理器，Sink Processors 类型包括这三种：Default Sink Processor、Load balancing Sink Processor 和 Failover Sink Processor。DefaultSink Processor 是默认的，不用配置 sink group，就是之前使用的形式，一个 channel 后面接一个 sink 的形式，Load balancing Sink Processor 是负载均衡处理器，一个 channle 后面可以接多个 sink，这多个 sink 属于一个 sink group，根据指定的算法进行轮询或者随机发送，减轻单个 sink 的压力，Failover Sink Processor 是故障转移处理器，一个 channle 后面可以接多个 sink，这多个 sink 属于一个 sink group，按照 sink 的优先级，默认先让优先级高的 sink 来处理数据，如果这个 sink 出现了故障，则用优先级低一点的 sink 处理数据，可以保证数据不丢失。

**先看负载均衡 Load balancing Sink Processor：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 39]  
看这个图，在这里面一个 channel 后面接了两个 sink，这两个 sink 属于一个 sink group，具体的数据发送策略我们来看一下文档：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 40]  
看中间的参数信息：

1.  processor.sinks：指定这个 sink groups 中有哪些 sink，指定 sink 的名称，多个的话中间使用空格隔开即可【注意，这里写的是 processor.sinks，但是在下面的 example 中使用的是 sinks，实际上就是 sinks，所以文档也是有一些瑕疵的，不过 Flume 的文档已经算是写的非常好的了】。
2.  processor.type：针对负载均衡的 sink 处理器，这里需要指定load\_balance
3.  processor.selector：此参数的值内置支持两个，round\_robin 和 random，round\_robin 表示轮询，按照 sink 的顺序，轮流处理数据，random 表示随机。
4.  processor.backoff：默认为 false，设置为 true 后，故障的节点会列入黑名单，过一定时间才会再次发送数据，如果还失败，则等待时间是指数级增长；一直到达到最大的时间。如果不开启，故障的节点每次还会被重试发送，如果真有故障节点的话就会影响效率。
5.  processor.selector.maxTimeOut：最大的黑名单时间，默认是30秒。

最后看一下里面提供的例子：

# 指定sinkgroup的名称
    a1.sinkgroups = g1
    # 指定sinkgroup中有哪些sink
    a1.sinkgroups.g1.sinks = k1 k2
    # 使用负载均衡策略
    a1.sinkgroups.g1.processor.type = load_balance
    # 开启后，故障的节点会列入黑名单
    a1.sinkgroups.g1.processor.backoff = true
    # 指定sinkgroup的数据发送策略
    a1.sinkgroups.g1.processor.selector = random

这个负载均衡案例可以解决之前单节点输出能力有限的问题，可以通过多个 sink 后面连接多个 Agent 实现负载均衡，如果后面的 Agent 挂掉1个，也不会影响整体流程，只是处理效率又恢复到了之前的状态。下面来配置 Agent，先配置 bigdata04 节点上的这个 Agent：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1 k2
    
    # 配置source组件
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 44444
    
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件,[为了方便演示效果，把batch-size设置为1]
    a1.sinks.k1.type=avro
    a1.sinks.k1.hostname=192.168.61.101
    a1.sinks.k1.port=41414
    a1.sinks.k1.batch-size = 1
    
    a1.sinks.k2.type=avro
    a1.sinks.k2.hostname=192.168.61.102
    a1.sinks.k2.port=41414
    a1.sinks.k2.batch-size = 1
    
    
    # 配置sink策略
    a1.sinkgroups = g1
    a1.sinkgroups.g1.sinks = k1 k2
    a1.sinkgroups.g1.processor.type = load_balance
    a1.sinkgroups.g1.processor.backoff = true
    a1.sinkgroups.g1.processor.selector = round_robin
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1

接着配置 bigdata02 上的 Agent，vi load-balancing-101.conf：

# agent的名称是a1
    
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 41414
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件[为了区分两个sink组件生成的文件，修改filePrefix的值]
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/load_balance
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    a1.sinks.k1.hdfs.filePrefix = data101
    a1.sinks.k1.hdfs.fileSuffix = .log
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

接着配置 bigdata03 上的 Agent，vi load-balancing-102.conf：

# agent的名称是a1
    
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 41414
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件[为了区分两个sink组件生成的文件，修改filePrefix的值]
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/load_balance
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    a1.sinks.k1.hdfs.filePrefix = data102
    a1.sinks.k1.hdfs.fileSuffix = .log
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

**注意：bigdata04 中指定的 a1.sinks.k1.port=41414 和 a1.sinks.k2.port=41414 需要和 bigdata02、bigdata03 中的 a1.sources.r1.port = 41414 的值保持一致**

接下来开始启动这几个 Agent，需要先启动 bigdata02 和 bigdata03 上的 Agent，最后启动 bigdata04 上的 Agent。**注意：在启动 bigdata02、bigdata03 之前需要到 /etc/profile 中先配置 HADOOP\_HOME 环境变量，因为 Agent 中使用到了 hdfs。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 41]  
通过 telnet 产生数据：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 42]  
分别到 bigdata02 和 bigdata03 中 hdfs sink 指定的 path 中验证数据：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 43]  
这就是 sink 负载均衡的使用，在这里使用了轮询的策略，读者可以使用 random 试验一下，只需要修改 bigdata04 配置文件中，a1.sinkgroups.g1.processor.selector 的值即可。

## 4.4 Failover Sink Processor ##

接下来看一下故障转移：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 44]  
在这个图中，也是一个 channel 后面接了两个 sink，但是这里和负载均衡架构不一样的是，这两个 sink 正常情况下只有一个干活，另一个是不干活的  
来看一下 Failover Sink Processor 的文档介绍：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 45]  
看中间的参数解释：

1.  sinks：指定这个 sink groups 中有哪些 sink，指定 sink 的名称，多个的话中间使用空格隔开即可。
2.  processor.type：针对故障转移的 sink 处理器，使用 failover。
3.  processor.priority.：指定 sink group 中每一个 sink 组件的优先级，默认情况下 channel 中的数据会被优先级比较高的 sink 取走。
4.  processor.maxpenalty：sink 发生故障之后，最大等待时间。

这个故障转移案例可以解决 sink 组件单点故障的问题，如果某一个 sink 输出功能失效，另一个还可以顶上来，同时只会存在一个真正输出数据的 sink。下面来配置 Agent，先配置 bigdata04 节点上的这个 Agent，vi failover.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1 k2
    
    # 配置source组件
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 44444
    
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    
    
    # 配置sink组件,[为了方便演示效果，把batch-size设置为1]
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = 192.168.61.101
    a1.sinks.k1.port = 41414
    a1.sinks.k1.batch-size = 1
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = 192.168.61.102
    a1.sinks.k2.port = 41414
    a1.sinks.k2.batch-size = 1
    
    
    
    # 配置sink策略
    a1.sinkgroups = g1
    a1.sinkgroups.g1.sinks = k1 k2
    a1.sinkgroups.g1.processor.type = failover
    a1.sinkgroups.g1.processor.priority.k1 = 5
    a1.sinkgroups.g1.processor.priority.k2 = 10
    a1.sinkgroups.g1.processor.maxpenalty = 10000
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1

接着配置 bigdata02 上的 Agent，vi failover-101.conf：

# agent的名称是a1
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 41414
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件[为了区分两个sink组件生成的文件，修改filePrefix的值]
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/failover
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    a1.sinks.k1.hdfs.filePrefix = data101
    a1.sinks.k1.hdfs.fileSuffix = .log
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

接着配置 bigdata03 上的 Agent，vi failover-102.conf：

# agent的名称是a1
    
    # 指定source组件、channel组件和Sink组件的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    # 配置source组件
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 41414
    
    # 配置channel组件
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 配置sink组件[为了区分两个sink组件生成的文件，修改filePrefix的值]
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.61.100:9000/failover
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    a1.sinks.k1.hdfs.filePrefix = data102
    a1.sinks.k1.hdfs.fileSuffix = .log
    
    
    # 把组件连接起来
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

接着就需要开始启动 Agent 了，先启动 bigdata02 和 bigdata03，最后启动 bigdata04。通过 telnet 产生数据：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 46]  
然后到 hdfs 上验证数据，发现现在数据是通过 bigdata03 这台机器写出去的，因为对应 bigdata03 这台机器的 sink 组件的优先级比较高：  
![在这里插入图片描述][ad74a01228a04f8e9f2ace7c8d853b9a.png]  
接下来模拟 bigdata03 这台机器上的 Agent 挂掉，也就意味着 k2 这个 sink 写不出去数据了，此时，我们再通过 socket 发送一条数据，看看会怎么样  
直接在 bigdata03 窗口中按 ctrl+c 停止 agent 即可，然后再生成一条数据，此时再到 hdfs 上查看一下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 47]  
查看新生成的这个文件的内容：  
![在这里插入图片描述][6249e0710fd14db49f7f0cbba8ac38e4.png]  
此时如果把 bigdata03 上的 Agent 再启动的话，会发现新采集的数据会通过 bigdata03 上的 Agent 写出去，这是因为它的优先级比较高。这就是 Sink 故障转移的应用。

# 五、Flume 出神入化篇 #

之前介绍了很多组件，有核心组件和高级组件：source、channel、sink 以及Source Interceptors，Channel Selectors、Sink Processors，针对这些组件，Flume 都内置提供了组件的很多具体实现，在实际工作中，95%以上的数据采集需求都是可以满足的，但是谁也不敢保证100%都能满足，因为什么奇葩的需求都会有，那针对系统内没有提供的一些组件怎么办呢？假设我们想把 flume 采集到的数据输出到 mysql 中，那这个时候就需要有针对 mysql 的 sink 组件了，但是 Flume 中并没有，因为这种需求不常见，往 mysql 中写的都是结构化数据，数据的格式是固定的，但是 flume 采集的一般都是日志数据，这种属于非结构化数据，不支持也是正常的，但是在这里就是需要使用 Flume 往 mysql 中写数据，那怎么办？要不我们考虑换一个采集工具把，当然这也是一种解决方案，如果有其他采集工具支持向 mysql 中写数据的话那可以考虑换一个采集工具，如果所有的采集工具都不支持向 mysql 中写数据呢，也就是说你这个需求就是前无古人后无来者的，怎么破？不用担心，天无绝人之路，其实之前使用的 Flume 提供的那些内置组件也都是作者一行代码一行代码写出来的，那我们是不是也可以自己写一个自定义的组件呢？可以的，并且 flume 也很欢迎你这样去做，它把开发文档什么的东西都给你准备好了。注意了，就算没有文档，我们也要想办法去自定义，没有文档的话就需要去抠 Flume 的源码了。在这里 Flume 针对自定义组件提供了详细的文档说明，我们来看一下，通过 Flume User Guide 可以看到，针对 source、channle、sink、Source Interceptors，Channel Selectors、都是可以的，这里面都显示了针对自定义的组件如何配置使用，Sink Processors 目前暂时不支持自定义。那这些支持自定义的组件具体开发步骤是什么样的呢？代码该写成什么样的呢？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 48]  
只不过开发者文档里面目前还不算太完善，但是基本 source、sink 组件的自定义过程在这里都是有的：  
![在这里插入图片描述][2353996a89d04299a7e5ad5b0e73631e.png]  
例如：自定义 source  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 49]  
例如：自定义 sink  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 50]  
自定义 channel 的内容目前还没完善，如果你确实想自定义这个组件，就需要到 Flume 源码中找到目前支持的那些 channel 的代码，参考着实现我们自定义的 channel 组件。大家在这里知道可以自定义，并且知道自定义组件的文档在哪里就可以了，目前来说，需要我们自定义组件的场景实在是太少了，几乎和买彩票中奖的概率差不多。

**Flume 优化：** 调整 Flume 进程的内存大小，建议设置 1G~2G，太小的话会导致频繁 GC。因为 Flume 进程也是基于 Java 的，所以就涉及到进程的内存设置，一般建议启动的单个 Flume 进程(或者说单个Agent)内存设置为 1G~2G，内存太小的话会频繁 GC，影响 Agent 的执行效率。那具体设置多少合适呢？这个需求需要根据 Agent 读取的数据量的大小和速度有关系，所以需要具体情况具体分析，当 Flume 的 Agent 启动之后，对应就会启动一个进程，我们可以通过 jstat -gcutil PID 1000 来看看这个进程 GC 的信息，每一秒钟刷新一次，如果 GC 次数增长过快，说明内存不够用。使用 jps 查看目前启动 flume 进程：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 51]  
在这里主要看 YGC YGCT FGC FGCT GCT

1.  YGC：表示新生代堆内存 GC 的次数，如果每隔几十秒产生一次，也还可以接受，如果每秒都会发生一次 YGC，那说明需要增加内存了。
2.  YGCT：表示新生代堆内存 GC 消耗的总时间。
3.  FGC：FULL GC 发生的次数，注意，如果发生 FUCC GC，则 Flume 进程会进入暂停状态，FUCC GC 执行完以后，Flume 才会继续工作，所以 FUCC GC 是非常影响效率的，这个指标的值越低越好，没有更好。
4.  GCT：所有类型的 GC 消耗的总时间。

如果需要调整 Flume 进程内存的话，需要调整 flume-env.sh 脚本中的 JAVA\_OPTS 参数，把 export JAVA\_OPTS 参数前面的\#号去掉才会生效。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 52]  
建议这里的 Xms 和 Xmx 设置为一样大，避免进行内存交换，内存交换也比较消耗性能。

在一台服务器启动多个 agent 的时候，建议修改配置区分日志文件，因为在 conf 目录下有 log4j.properties，在这里面指定了日志文件的名称和位置，所有使用 conf 目录下面配置启动的 Agent 产生的日志都会记录到同一个日志文件中，如果我们在一台机器上启动了10几个 Agent，后期发现某一个 Agent 挂了，想要查看日志分析问题，这个时候就疯了，因为所有 Agent 产生的日志都混到一块了，压根都没法分析日志了。所以建议拷贝多个 conf 目录，然后修改对应 conf 目录中 log4j.properties 日志的文件名称(可以保证多个agent的日志分别存储)，并且把日志级别调整为warn(减少垃圾日志的产生)，默认 info 级别会记录很多日志信息。这样在启动 Agent 的时候分别通过 –conf 参数指定不同的 conf 目录，后期分析日志就方便了，每一个 Agent 都有一个单独的日志文件。以 bigdata04 机器为例：复制 conf-failover 目录，以后启动 sink 的 failover 任务的时候使用这个目录，修改 log4j.properties 中的日志记录级别和日志文件名称，日志文件目录可以不用修改，统一使用 logs 目录即可。cd /data/soft/apache-flume-1.9.0-bin、cp -r conf/ conf-failover、cd conf-failover/、vi log4j.properties：  
![在这里插入图片描述][d4fadf92372843c4b43d177f9d1a79d2.png]  
再启动的时候就是这样的了：

nohup bin/flume-ng agent --name a1 --conf conf-failover --conf-file conf/failover.conf &

**Flume 进程监控：** Flume 的 Agent 服务是一个独立的进程，假设我们使用 source->channel->sink 实现了一个数据采集落盘的功能，如果这个采集进程被误操作干掉了，这个时候我们是发现不了的，什么时候会发现呢？可能第二天，产品经理找到你了，说昨天的这个指标值有点偏低啊，你来看下怎么回事，然后你就一顿操作猛如虎，结果发现原始数据少了一半多，那是因为 Flume 的采集程序在昨天下午的时候被误操作干掉了。找到问题之后，你就苦巴巴的手工去补数据，重跑计算程序，最后再找产品经理确认数据的准确性。类似的问题会有很多，这说明你现在是无法掌控你手下的这些程序，他们都是不受控的状态，说不定哪天哪个程序不高兴，他就自杀了，不干活了，过了好几天，需要用到这个数据的时候你才发现，发现的早的话还能补数据，发现晚的话数据可能都补不回来了，这样对公司来说就是属于比较严重的数据故障问题，这样你年终奖想拿18薪就不太现实了。所以针对这些存在单点故障的进程，我们都需要添加监控告警机制，最起码出问题能及时知道，再好一点的呢，可以尝试自动修复重启。那针对 Flume 中的 Agent 我们就来实现一个监控功能，并且尝试自动重启，大致思路是这样的，首先需要有一个配置文件，配置文件中指定你现在需要监控哪些 Agent，有一个脚本负责读取配置文件中的内容，定时挨个检查 Agent 对应的进程还在不在，如果发现对应的进程不在，则记录错误信息，然后告警(发短信或者发邮件) 并尝试重启，创建一个文件 monlist.conf ，文件中的第一列指定一个 Agent 的唯一标识，后期需要根据这个标识过滤对应的 Flume 进程，所以一定要保证至少在一台机器上是唯一的，等号后面是一个启动 Flume 进程的脚本，这个脚本和 Agent 的唯一标识是一一对应的，后期如果根据 Agent 标识没有找到对应的进程，那么就需要根据这个脚本启动进程：

example=startExample.sh

这个脚本的内容如下：startExample.sh：

#!/bin/bash
    flume_path=/data/soft/apache-flume-1.9.0-bin
    nohup ${flume_path}/bin/flume-ng agent --name a1 --conf ${flume_path}/conf/ --conf-file ${flume_path}/conf/example.conf  &

接着就是要写一个脚本来检查进程在不在，不在的话尝试重启，创建脚本 monlist.sh：

#!/bin/bash
    monlist=`cat monlist.conf`
    echo "start check"
    for item in ${monlist}
    do
    	# 设置字段分隔符
    	OLD_IFS=$IFS
    	IFS="="
    	# 把一行内容转成多列[数组]
    	arr=($item)
    	# 获取等号左边的内容
    	name=${arr[0]}
    	# 获取等号右边的内容
    	script=${arr[1]}
    	
    	echo "time is:"`date +"%Y-%m-%d %H:%M:%S"`" check "$name
    	if [ `jps -m|grep $name | wc -l` -eq 0 ]
    	then
    		# 发短信或者邮件告警
    		echo `date +"%Y-%m-%d %H:%M:%S"`$name "is none"
    		sh -x ./${script}
    	fi
    done

注意：这个需要定时执行，所以可以使用 crontab 定时调度。

* * * * *  root /bin/bash /data/soft/monlist.sh

部分笔记参考徐葳(作者)老师！！！！猎豹移动大数据技术专家，中科院大数据研究院大数据技术专家，专注于大数据技术开发与研究，拥有8年一线互联网公司软件研发经验，曾为移动研究院、中移在线、中国联通、新华三等企业进行大数据技术企培，著有《Flink入门与实战》一书。

[2021_ _ Hadoop]: https://blog.csdn.net/xw1680/article/details/115426714
[2021_ _Hadoop]: https://blog.csdn.net/xw1680/article/details/115434353
[2021_ _Hadoop 1]: https://blog.csdn.net/xw1680/article/details/115500104
[2021_ _Hadoop _ HDFS]: https://blog.csdn.net/xw1680/article/details/115555808
[2021_ _Zookeeper]: https://blog.csdn.net/xw1680/article/details/118002073
[2021_ _Hadoop _ HDFS 1]: https://editor.csdn.net/md/?articleId=118033171
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70]: /images/20220829/dba41d761850409fb2aaff38b3117563.png
[20210702003423927.png]: /images/20220829/db056638ffb540e8b2c494b6db12ba85.png
[d060aef43e8d4316babd2704b3f45dd9.png]: /images/20220829/2f698e801181401b830c656ef37bf7fd.png
[20210702010323128.png]: /images/20220829/b1d006bcf66f4d5cbd502dae96d3b07d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 1]: /images/20220829/96cf67dc5bb74cb8bc03f0988cc4ee9a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 2]: /images/20220829/adcc4b41a6994ac29bb68440c5b6f224.png
[20210702011309461.png]: /images/20220829/676235ea116a4eb7bdf6b5ceed247a1a.png
[Link 1]: http://flume.apache.org/download.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 3]: /images/20220829/9b5da0c95bec4f58a1da848cfab6ab4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 4]: /images/20220829/308c502ac0514088bf1d6581154b70e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 5]: /images/20220829/85a6d2570dd8468fa4c49973f986dad7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 6]: /images/20220829/b731d6a5760540c9bf298e6ad8929a9e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 7]: /images/20220829/689fadd3e7654d34ae07c4931046bfb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 8]: /images/20220829/22908afa3a784e7fbb814f992bd3f7cc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 9]: /images/20220829/6d972ab7d0f94bb4a1681eeff41564b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 10]: /images/20220829/ca780a786ade42f29d979682787a07c8.png
[20210703135720198.png]: /images/20220829/244f41f9b3a4429da995ec63a69373f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 11]: /images/20220829/3994887e6ea942f4927e427d4944d771.png
[20210703140953286.png]: /images/20220829/3a0fce2fef914f97bf156e649ac7c490.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 12]: /images/20220829/82c4262a1f3e422c8bbc947fee8c5037.png
[20210703141345902.png]: /images/20220829/0f3d45cd76bd4b3792bf2e0894a2cd8d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 13]: /images/20220829/10e0f52ecf604de5b26358b2a6b07d2c.png
[20210703143159472.png]: /images/20220829/9d49ddaf8259452d9fa6fe040e0af817.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 14]: /images/20220829/7d523c9c2af1439ba2cff263a5615ab1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 15]: /images/20220829/66e837bd99c5442398e9c88207dfa9a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 16]: /images/20220829/b5acfa96d0b94f79aff21e588166e528.png
[20210703144736119.png]: /images/20220829/2a25e2697ffb47cea431056116865395.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 17]: /images/20220829/448af8ab900649f4a5ab722df3d41bd0.png
[20210703145308190.png]: /images/20220829/37dee44e189f4dcf979419479c3a69f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 18]: /images/20220829/a166af40acb34214840ac3be49f3968a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 19]: /images/20220829/ba2ca9665b56463cbf0bc5f22f1fa652.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 20]: /images/20220829/2353489d6f2a4959ae7cb1edf92abd8f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 21]: /images/20220829/6f545f89f3214e55a1b074062878a0c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 22]: /images/20220829/a61988afd0d5449b84dbd2b2e057bc21.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 23]: /images/20220829/7e25ea9d54264b0387e988eadc2d4695.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 24]: /images/20220829/d6d9af5f9a6b4e7a8fdea8390ef598c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 25]: /images/20220829/d41cb25f229f4f2d9bf34be168215665.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 26]: /images/20220829/4a9989b058a6491090bcc24f19bbdad6.png
[20210703185910657.png]: /images/20220829/84d3f26cc4094bc8ae6d351523eeaf2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 27]: /images/20220829/ec7b8ed2c70a4d5e8cba698b57d6fdf0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 28]: /images/20220829/daf3a10acfbe4b14abb852f886232d46.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 29]: /images/20220829/af39da8bf40a4727a8fb78789b9965e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 30]: /images/20220829/f50e6a6c9a03403a94d727cfc41dcf43.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 31]: /images/20220829/11aea83ef28543bab42d2808d2a20016.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 32]: /images/20220829/03be99f3b6194d3da7dd32d5e6f6ed49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 33]: /images/20220829/4b461c5c520342cbaf5c8c816ae197c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 34]: /images/20220829/33f47dd7213f4b169cbdc9153149e285.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 35]: /images/20220829/8c5d3f0acaf7408f8d618dac91cca709.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 36]: /images/20220829/62f8b5d5675a4675a4cd562e6dca80ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 37]: /images/20220829/5e18eb3210db4a1cb7b7bdc5e41dbc60.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 38]: /images/20220829/fc9b245ffb144ea5868e75c7d8436c66.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 39]: /images/20220829/45347d5ec6124829a8fe8c9f01964e13.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 40]: /images/20220829/91308db05f624bd6b88b843007f76eeb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 41]: /images/20220829/0ddb2068719742b7b5a02102775253eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 42]: /images/20220829/ceebcb5f9f8740cea2225b7705d14e44.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 43]: /images/20220829/6dcb872b4c134ed4a4b39a3c4745d9f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 44]: /images/20220829/e68736f6d26642edb4496ef0f4951f7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 45]: /images/20220829/11b5b2d0866e45c9984328be05ecd785.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 46]: /images/20220829/daa450ce3cc4482a8f5663618dbda387.png
[ad74a01228a04f8e9f2ace7c8d853b9a.png]: /images/20220829/426b8086d00b4fa1a71eccf0fd1c866d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 47]: /images/20220829/23534ab033244695b1d1f1e851cdb5cb.png
[6249e0710fd14db49f7f0cbba8ac38e4.png]: /images/20220829/0ff69f804ed54a22a4d9839105b44e5f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 48]: /images/20220829/3449b82f5b4943c98414bb3af92a5e5d.png
[2353996a89d04299a7e5ad5b0e73631e.png]: /images/20220829/e1bc9f08a896434e8030962af93e3b52.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 49]: /images/20220829/58086b3bf37d4316a01403f2579b4216.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 50]: /images/20220829/4893fc82c6cb4e6ea301feeebd75dcd1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 51]: /images/20220829/5cfe50b1fd6542b8a0bec382c4db360d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h3MTY4MA_size_16_color_FFFFFF_t_70 52]: /images/20220829/e70a426b37924c50a743a6ecbec7f09d.png
[d4fadf92372843c4b43d177f9d1a79d2.png]: /images/20220829/d8aa42c3ca8943c3b9a40c28c44a5e44.png