大数据必学框架-Flume

秒速五厘米 2023-10-04 16:41 25阅读 0赞

## 大数据必学框架-Flume ##

### Flume简介 ###

#### Flume定义： ####

Flume 是 Cloudera 提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统。Flume 基于流式架构，灵活简单。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70]

\*\*官网地址：\*\*https://flume.apache.org

#### Flume的基础架构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 1]

介绍一下Flume组件

##### Agent #####

Agent 是一个 JVM 进程，它以事件的形式将数据从源头送至目的。  
Agent 主要有 3 个部分组成，**Source、Channel、Sink**。

##### Source #####

Source 是负责接收数据到 Flume Agent 的组件。Source 组件可以处理各种类型、各种格式的日志数据，包括 avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。

##### Sink #####

Sink 不断地轮询 Channel 中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。  
Sink 组件目的地包括 hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定  
义。

##### Channel #####

Channel 是位于 Source 和 Sink 之间的缓冲区。因此，Channel 允许 Source 和 Sink 运作在不同的速率上。Channel 是线程安全的，可以同时处理几个 Source 的写入操作和几个Sink 的读取操作。  
Flume 自带两种 Channel：Memory Channel 和 File Channel 以及 Kafka Channel。  
Memory Channel 是内存中的队列。Memory Channel 在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么 Memory Channel 就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。  
File Channel 将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数  
据。

##### Event #####

传输单元，Flume 数据传输的基本单元，以 Event 的形式将数据从源头送至目的地。  
Event 由 Header 和 Body 两部分组成，Header 用来存放该 event 的一些属性，为 K-V 结构，Body 用来存放该条数据，形式为字节数组。

### Flume快速入门 ###

#### Flume的安装部署 ####

##### Flume下载地址 #####

https://flume.apache.org/download.html

##### 文档查看地址 #####

https://flume.apache.org/FlumeUserGuide.html

##### 安装Flume #####

**Flume和ZooKeeper一样，都是由java开发的，所以在安装之前要确定安装过了java**

下载完得tar包，解压之后放到相应的文件夹后，打开flume的配置文件夹conf,会看到一个**flume-env.sh.template**文件，这是官方给出的参考模板，可以对其修改名字，然后进去修改配置

flume的配置十分简单，只需要**修改JAVA\_HOME的路径**即可

#### Flume的入门案例 ####

##### 需求：使用Flume监听一个端口，收集该端口数据，并打印到控制台 #####

分析：监听一个端口需要使用netcat工具，输出到控制台则是logger  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 2]

**编写配置文件flume-netcat-logger.conf**

添加内容如下： 
    # 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

**开启flume监听端口**

两种写法

**第一种**

bin/flume-ng agent --conf conf/ --name a1  --conf-file  job/flume-netcat-logger.conf  -Dflume.root.logger=INFO,console

**第二种**

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

**参数说明**

–conf/-c：表示配置文件存储在 conf/目录

–name/-n：表示给 agent 起名为 a1

–conf-file/-f：flume 本次启动读取的配置文件是在 job 文件夹下的 flume-telnet.conf

##### 实时 监控单个追加 文件 #####

\*\*案例需求：实时监控 H ive 日志，并上传到 HDFS 中 \*\*

**分析：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 3]

Source：监听文件，使用**exec**，exec 即 execute 执行的意思。表示执行 Linux  
命令来读取文件

Sink：传输到HDFS，使用hdfs

Channel：依旧使用memory

**实现步骤：**

Flume 要想将数据输出到 HDFS ，须持有 Hadoop 相关 jar 包

commons-configuration-1.6.jar、 
    hadoop-auth-2.7.2.jar、 
    hadoop-common-2.7.2.jar、 
    hadoop-hdfs-2.7.2.jar、 
    commons-io-2.4.jar、 
    htrace-core-3.1.0-incubating.jar

拷贝到/opt/module/flume/lib 文件夹下。

\*\*创建 flume- file- hdfs.conf 文件 \*\*

# Name the components on this agent 
    a2.sources = r2 
    a2.sinks = k2 
    a2.channels = c2 
     
    # Describe/configure the source 
    a2.sources.r2.type = exec 
    a2.sources.r2.command = tail -F /opt/module/hive/logs/hive.log 
    a2.sources.r2.shell = /bin/bash -c 
     
    # Describe the sink 
    a2.sinks.k2.type = hdfs 
    a2.sinks.k2.hdfs.path = hdfs://hadoop102:9000/flume/%Y%m%d/%H 
    #上传文件的前缀 
    a2.sinks.k2.hdfs.filePrefix = logs- 
    #是否按照时间滚动文件夹 
    a2.sinks.k2.hdfs.round = true 
    #多少时间单位创建一个新的文件夹 
    a2.sinks.k2.hdfs.roundValue = 1 
    #重新定义时间单位 
    a2.sinks.k2.hdfs.roundUnit = hour 
    #是否使用本地时间戳 
    a2.sinks.k2.hdfs.useLocalTimeStamp = true 
    #积攒多少个 Event 才 flush 到 HDFS 一次 
    a2.sinks.k2.hdfs.batchSize = 1000 
    #设置文件类型，可支持压缩 
    a2.sinks.k2.hdfs.fileType = DataStream 
    #多久生成一个新的文件 
    a2.sinks.k2.hdfs.rollInterval = 30 
    #设置每个文件的滚动大小 
    a2.sinks.k2.hdfs.rollSize = 134217700 
    #文件的滚动与 Event 数量无关 
    a2.sinks.k2.hdfs.rollCount = 0 
     
    # Use a channel which buffers events in memory 
    a2.channels.c2.type = memory 
    a2.channels.c2.capacity = 1000 
    a2.channels.c2.transactionCapacity = 100 
     
    # Bind the source and sink to the channel 
    a2.sources.r2.channels = c2 
    a2.sinks.k2.channel = c2

##### 实时监控 目录 下多个 新 文件 #####

\*\*案例需求：使用 F lume 监听整个目录的文件 ，并上传至 HDFS \*\*

**分析：**

Source：监控整个目录：Spooldir

Sink：上传到HDFS：hdfs

Channel：使用memory  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 4]

**实现步骤：**

**创建配置文件 flume- dir- hdfs.conf**

a3.sources = r3 
    a3.sinks = k3 
    a3.channels = c3 
     
    # Describe/configure the source 
    a3.sources.r3.type = spooldir 
    a3.sources.r3.spoolDir = /opt/module/flume/upload 
    a3.sources.r3.fileSuffix = .COMPLETED 
    a3.sources.r3.fileHeader = true 
    #忽略所有以.tmp 结尾的文件，不上传 
    a3.sources.r3.ignorePattern = ([^ ]*\.tmp) 
     
    # Describe the sink 
    a3.sinks.k3.type = hdfs 
    a3.sinks.k3.hdfs.path  = 
    hdfs://hadoop102:9000/flume/upload/%Y%m%d/%H 
    #上传文件的前缀 
    a3.sinks.k3.hdfs.filePrefix = upload- 
    #是否按照时间滚动文件夹 
    a3.sinks.k3.hdfs.round = true 
    #多少时间单位创建一个新的文件夹 
    a3.sinks.k3.hdfs.roundValue = 1 
    #重新定义时间单位 
    a3.sinks.k3.hdfs.roundUnit = hour 
    #是否使用本地时间戳 
    a3.sinks.k3.hdfs.useLocalTimeStamp = true 
    #积攒多少个 Event 才 flush 到 HDFS 一次 
    a3.sinks.k3.hdfs.batchSize = 100 
    #设置文件类型，可支持压缩 
    a3.sinks.k3.hdfs.fileType = DataStream 
    #多久生成一个新的文件 
    a3.sinks.k3.hdfs.rollInterval = 60 
    #设置每个文件的滚动大小大概是 128M 
    a3.sinks.k3.hdfs.rollSize = 134217700 
    #文件的滚动与 Event 数量无关 
    a3.sinks.k3.hdfs.rollCount = 0 
     
    # Use a channel which buffers events in memory 
    a3.channels.c3.type = memory 
    a3.channels.c3.capacity = 1000 
    a3.channels.c3.transactionCapacity = 100 
     
    # Bind the source and sink to the channel 
    a3.sources.r3.channels = c3 
    a3.sinks.k3.channel = c3

##### 实时监控目录下的多个追加文件 #####

Exec source 适用于监控一个实时追加的文件，但不能保证数据不丢失；Spooldir  
Source 能够保证数据不丢失， 且能够实现断点续传， 但延迟较高， 不能实时监控； 而 **Taildir Source 既能够实现断点续传**，又可以保证数据不丢失，还能够进行实时监控。

**案例需求：使用Flume监听整个目录的实时追加文件，并且上传至HDFS**

**分析：**

Source：实时追加监听：T**AILDIR：支持断点续传，而exec不支持断点续传**

Sink：上传至HDFS：hdfs

Channel：使用memory  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 5]

**实现步骤**

\*\*创建配置文件 flume- tail dir- hdfs.conf \*\*

a3.sources = r3 
    a3.sinks = k3 
    a3.channels = c3 
     
    # Describe/configure the source 
    a3.sources.r3.type = TAILDIR 
    a3.sources.r3.positionFile = /opt/module/flume/tail_dir.json 
    a3.sources.r3.filegroups = f1 
    a3.sources.r3.filegroups.f1 = /opt/module/flume/files/file.* 
     
    # Describe the sink 
    a3.sinks.k3.type = hdfs 
    a3.sinks.k3.hdfs.path  = 
    hdfs://hadoop102:9000/flume/upload/%Y%m%d/%H 
    #上传文件的前缀 
    a3.sinks.k3.hdfs.filePrefix = upload- 
    #是否按照时间滚动文件夹 
    a3.sinks.k3.hdfs.round = true 
    #多少时间单位创建一个新的文件夹 
    a3.sinks.k3.hdfs.roundValue = 1 
    #重新定义时间单位 
    a3.sinks.k3.hdfs.roundUnit = hour 
    #是否使用本地时间戳 
    a3.sinks.k3.hdfs.useLocalTimeStamp = true 
    #积攒多少个 Event 才 flush 到 HDFS 一次 
    a3.sinks.k3.hdfs.batchSize = 100 
    #设置文件类型，可支持压缩 
    a3.sinks.k3.hdfs.fileType = DataStream 
    #多久生成一个新的文件 
    a3.sinks.k3.hdfs.rollInterval = 60 
    #设置每个文件的滚动大小大概是 128M 
    a3.sinks.k3.hdfs.rollSize = 134217700 
    #文件的滚动与 Event 数量无关 
    a3.sinks.k3.hdfs.rollCount = 0 
     
    # Use a channel which buffers events in memory 
    a3.channels.c3.type = memory 
    a3.channels.c3.capacity = 1000 
    a3.channels.c3.transactionCapacity = 100 
     
    # Bind the source and sink to the channel 
    a3.sources.r3.channels = c3 
    a3.sinks.k3.channel = c3

**Taildir 说明：**

Taildir Source 维护了一个 json 格式的 position File， 其会定期的往 position File中更新每个文件读取到的最新的位置， 因此能够实现断点续传。 Position File的格式下：

{
        "inode":2496272,"pos":12,"file":"/opt/module/flume/files/file1.txt"} 
    {
        "inode":2496275,"pos":12,"file":"/opt/module/flume/files/file2.txt"}

### Flume进阶 ###

#### Flume事务 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 6]

#### Flume Agent内部原理 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 7]

#### 重要组件： ####

##### ChannelSelector #####

ChannelSelector 的作用就是选出 Event 将要被发往哪个 Channel。其共有两种类型，分别是 Replicating（复制）和 Multiplexing（多路复用） 。

ReplicatingSelector 会将同一个 Event 发往所有的 Channel，Multiplexing 会根据相  
应的原则，将不同的 Event 发往不同的 Channel。

##### SinkProcessor #####

SinkProcessor 共 有 三 种 类 型 ， 分 别 是 DefaultSinkProcessor 、LoadBalancingSinkProcessor 和 FailoverSinkProcessor

DefaultSinkProcessor 对应的是单个的 Sink，LoadBalancingSinkProcessor 和  
FailoverSinkProcessor 对应的是 Sink Group，LoadBalancingSinkProcessor 可以实现负载均衡的功能，FailoverSinkProcessor 可以实现故障转移的功能。

### Flume拓扑结构 ###

#### 简单串联 ####

前一个Flume的输出是后一个Flume的输入

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 8]

#### 复制和多路复用 ####

一个source，多个channel和sink

Flume 支持将事件流向一个或者多个目的地。这种模式可以将相同数据复制到多个  
channel 中，或者将不同数据分发到不同的 channel 中，sink 可以选择传送到不同的目的地。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 9]

#### 负载均衡和故障转移 ####

Flume支持使用将多个sink逻辑上分到一个sink组， sink组配合不同的SinkProcessor  
可以实现负载均衡和错误恢复的功能。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 10]

#### 聚合 ####

这种模式是我们最常见的，也非常实用，日常 web 应用通常分布在上百个服务器，大者甚至上千个、上万个服务器。产生的日志，处理起来也非常麻烦。用 flume 的这种组合方式能很好的解决这一问题， 每台服务器部署一个 flume 采集日志， 传送到一个集中收集日志的flume，再由此 flume 上传到 hdfs、hive、hbase 等，进行日志分析。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 11]

### Flume企业开发案例 ###

#### 复制和多路复用 ####

**案例需求**

使用 Flume-1 监控文件变动，Flume-1 将变动内容传递给 Flume-2，Flume-2 负责存储到HDFS。 同时Flume-1将变动内容传递给Flume-3， Flume-3负责输出到Local FileSystem。

**需求分析**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 12]

**实现步骤**

创建flume-file-flume.conf

配置 1 个接收日志文件的 source 和两个 channel、两个 sink，分别输送给 flume-flume-hdfs 和 flume-flume-dir。

# Name the components on this agent 
    a1.sources = r1 
    a1.sinks = k1 k2 
    a1.channels = c1 c2 
     
    # 将数据流复制给所有 channel 
    a1.sources.r1.selector.type = replicating 
     
    # Describe/configure the source 
    a1.sources.r1.type = exec 
    a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log 
    a1.sources.r1.shell = /bin/bash -c 
     
    # Describe the sink 
    # sink 端的 avro 是一个数据发送者 
    a1.sinks.k1.type = avro 
    a1.sinks.k1.hostname = hadoop102 
    a1.sinks.k1.port = 4141 
     
    a1.sinks.k2.type = avro 
    a1.sinks.k2.hostname = hadoop102 
    a1.sinks.k2.port = 4142 
     
    # Describe the 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 
     
    # Bind the source and sink to the channel 
    a1.sources.r1.channels = c1 c2 
    a1.sinks.k1.channel = c1 
    a1.sinks.k2.channel = c2

创建 flume- flume- hdfs .conf

配置上级 Flume 输出的 Source，输出是到 HDFS 的 Sink。

# Name the components on this agent 
    a2.sources = r1 
    a2.sinks = k1 
    a2.channels = c1 
     
    # Describe/configure the source 
    # source 端的 avro 是一个数据接收服务 
    a2.sources.r1.type = avro 
    a2.sources.r1.bind = hadoop102 
    a2.sources.r1.port = 4141 
     
    # Describe the sink 
    a2.sinks.k1.type = hdfs 
    a2.sinks.k1.hdfs.path = hdfs://hadoop102:9000/flume2/%Y%m%d/%H 
    #上传文件的前缀 
    a2.sinks.k1.hdfs.filePrefix = flume2- 
    #是否按照时间滚动文件夹 
    a2.sinks.k1.hdfs.round = true 
    #多少时间单位创建一个新的文件夹 
    a2.sinks.k1.hdfs.roundValue = 1 
    #重新定义时间单位 
    a2.sinks.k1.hdfs.roundUnit = hour 
    #是否使用本地时间戳 
    a2.sinks.k1.hdfs.useLocalTimeStamp = true 
    #积攒多少个 Event 才 flush 到 HDFS 一次 
    a2.sinks.k1.hdfs.batchSize = 100 
    #设置文件类型，可支持压缩 
    a2.sinks.k1.hdfs.fileType = DataStream 
    #多久生成一个新的文件 
    a2.sinks.k1.hdfs.rollInterval = 600 
    #设置每个文件的滚动大小大概是 128M 
    a2.sinks.k1.hdfs.rollSize = 134217700 
    #文件的滚动与 Event 数量无关 
    a2.sinks.k1.hdfs.rollCount = 0 
     
    # Describe the channel 
    a2.channels.c1.type = memory 
    a2.channels.c1.capacity = 1000 
    a2.channels.c1.transactionCapacity = 100 
     
    # Bind the source and sink to the channel 
    a2.sources.r1.channels = c1 
    a2.sinks.k1.channel = c1

创建 flume- flume- dir .conf

配置上级 Flume 输出的 Source，输出是到本地目录的 Sink。

# Name the components on this agent 
    a3.sources = r1 
    a3.sinks = k1 
    a3.channels = c2 
     
    # Describe/configure the source 
    a3.sources.r1.type = avro 
    a3.sources.r1.bind = hadoop102 
    a3.sources.r1.port = 4142 
    
    # Describe the sink 
    a3.sinks.k1.type = file_roll 
    a3.sinks.k1.sink.directory = /opt/module/data/flume3 
     
    # Describe the channel 
    a3.channels.c2.type = memory 
    a3.channels.c2.capacity = 1000 
    a3.channels.c2.transactionCapacity = 100 
     
    # Bind the source and sink to the channel 
    a3.sources.r1.channels = c2 
    a3.sinks.k1.channel = c2

#### 自定义 Interceptor ####

**案例需求**

使用 Flume 采集服务器本地日志， 需要按照日志类型的不同， 将不同种类的日志发往不同的分析系统。

**需求分析**

在实际的开发中， 一台服务器产生的日志类型可能有很多种， 不同类型的日志可能需要发送到不同的分析系统。 此时会用到 Flume 拓扑结构中的 Multiplexing 结构， Multiplexing的原理是，根据 event 中 Header 的某个 key 的值，将不同的 event 发送到不同的 Channel中，所以我们需要自定义一个 Interceptor，为不同类型的 event 的 Header 中的 key 赋予不同的值。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 13]

**实现步骤**

新建maven工程，导入依赖

<dependency> 
        <groupId>org.apache.flume</groupId> 
        <artifactId>flume-ng-core</artifactId> 
        <version>1.7.0</version> 
    </dependency>

**自定义的拦截器需要实现Interceptor接口，然后重写四个方法**

//初始化方法
    public void initialize(){
        } 
    
    //处理单个事件，Event中封装了Header和Body，可以根据body内容设置为响应的//Header，然后Flume会以Header进行区分
    public Event intercept(Event event){
        }
    
    //处理多个事件，可以调用处理单个事件的intercept()
    public List<Event> intercept(List<Event> list){
        }
    
    public void close();
    
    //注意：还要定义一个静态内部类，实现Interceptor.Builder,在实现的build方法中，返回一个本类实例

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 14]

package com.gis.interceptor;
    
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    
    /**
     * @author LnnuUser
     * @create 2021-08-03-下午8:16
     */
    public class TypeInterceptor implements Interceptor {
        
    
        //声明一个存放事件的集合
        private ArrayList<Event> addHeaderEvents;
    
        @Override
        public void initialize() {
        
    
            //初始化存放事件的集合
            addHeaderEvents = new ArrayList<>();
        }
    
        /**
         * 处理单个事件
         * @param event
         * @return
         */
        @Override
        public Event intercept(Event event) {
        
    
            //1、获取头信息
            Map<String, String> headers = event.getHeaders();
    
            //2、获取body
            String body = new String(event.getBody());
    
            //3、根据body中是否有“hello”来决定添加怎样的头信息
            if (body.contains("hello")){
        
    
                //4、添加头信息
                headers.put("type","GIS");
            }else {
        
    
                //4、添加头信息
                headers.put("type","AIS");
            }
    
            return event;
        }
    
        @Override
        public List<Event> intercept(List<Event> list) {
        
    
            //1、清空集合
            addHeaderEvents.clear();
    
            //2、遍历events，给每一个事件添加头信息
            for (Event event : list) {
        
    
                //3、给每一个事件添加头信息
                addHeaderEvents.add(intercept(event));
    
            }
    
            //4、返回结果
            return addHeaderEvents;
        }
    
        @Override
        public void close() {
        
    
        }
    
        public static class Builder implements Interceptor.Builder{
        
    
            @Override
            public Interceptor build() {
        
    
                return new TypeInterceptor();
            }
    
            @Override
            public void configure(Context context) {
        
    
            }
        }
    }

\*\*编辑 flume 配置文件 \*\*

添加拦截器，拦截器就是自己写的全类名，要**打包放到flume的lib目录中**

a1.sources.r1.interceptors = i1 
    a1.sources.r1.interceptors.i1.type  = 
    com.gis.flume.interceptor.CustomInterceptor$Builder 
    a1.sources.r1.selector.type = multiplexing 
    
    a1.sources.r1.selector.header = type 
    a1.sources.r1.selector.mapping.letter = c1 
    a1.sources.r1.selector.mapping.number = c2

全部的配置

# Name the components on this agent 
    a1.sources = r1 
    a1.sinks = k1 k2 
    a1.channels = c1 c2 
     
    # Describe/configure the source 
    a1.sources.r1.type = netcat 
    a1.sources.r1.bind = localhost 
    a1.sources.r1.port = 44444 
     
    a1.sources.r1.interceptors = i1 
    a1.sources.r1.interceptors.i1.type  = 
    com.gis.flume.interceptor.CustomInterceptor$Builder 
    a1.sources.r1.selector.type = multiplexing 
    
    a1.sources.r1.selector.header = type 
    a1.sources.r1.selector.mapping.letter = c1 
    a1.sources.r1.selector.mapping.number = c2 
     
    # Describe the sink 
    a1.sinks.k1.type = avro 
    a1.sinks.k1.hostname = hadoop103 
    a1.sinks.k1.port = 4141 
     
    a1.sinks.k2.type=avro 
    a1.sinks.k2.hostname = hadoop104 
    a1.sinks.k2.port = 4242 
     
    # Use a channel which buffers events in memory 
    a1.channels.c1.type = memory 
    a1.channels.c1.capacity = 1000 
    a1.channels.c1.transactionCapacity = 100 
     
    # Use a channel which buffers events in memory 
    a1.channels.c2.type = memory 
    a1.channels.c2.capacity = 1000 
    a1.channels.c2.transactionCapacity = 100 
     
     
    # Bind the source and sink to the channel 
    a1.sources.r1.channels = c1 c2 
    a1.sinks.k1.channel = c1 
    a1.sinks.k2.channel = c2

#### 自定义 Source ####

Source 是负责接收数据到 Flume Agent 的组件。Source 组件可以处理各种类型、各种格式的日志数据，包括 avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。官方提供的 source 类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些 source。  
官方也提供了自定义 source 的接口：  
https://flume.apache.org/FlumeDeveloperGuide.html\#source 根据官方说明自定义  
MySource 需要继承 AbstractSource 类并实现 Configurable 和 PollableSource 接口。

**实现步骤**

自定义类继承AbstractSource并且实现Configurable、PollableSource，然后重写四个方法

//配置方法，从中读取配置文件中的信息
    public void configure(Context context) {
        }
    
    //主要的方法，逻辑都写在该方法中，返回值为Status状态，执行成功Status.READY,执行失败Status.BACKOFF
    public Status process(){
        }
    
    //这两个方法很少用到
    @Override
    public long getBackOffSleepIncrement() {
        
        return 0;
    }
    
    @Override
    public long getMaxBackOffSleepInterval() {
        
        return 0;
    }

代码实现

package com.gis.source;
    
    import org.apache.flume.Context;
    import org.apache.flume.EventDeliveryException;
    import org.apache.flume.PollableSource;
    import org.apache.flume.conf.Configurable;
    import org.apache.flume.event.SimpleEvent;
    import org.apache.flume.source.AbstractSource;
    
    import java.nio.charset.StandardCharsets;
    
    /**
     * @author LnnuUser
     * @create 2021-08-04-上午9:19
     */
    public class MySource extends AbstractSource implements Configurable, PollableSource {
        
    
        //定义全局的前缀和后缀
        private String prefix;
        private String subfix;
    
        Status status = null;
    
        /**
         *
         * @param context
         */
        @Override
        public void configure(Context context) {
        
    
            //读取配置信息，给前后缀赋值
            prefix = context.getString("prefix");
            subfix = context.getString("subfix","gis");
        }
    
        /**
         * 1.接受数据（for循环造数据）
         * 2.封装事件
         * 3.将事件传给channel
         *
         * @return
         * @throws EventDeliveryException
         */
        @Override
        public Status process() throws EventDeliveryException {
        
    
            try {
        
                //1.接受数据
                for (int i = 0; i < 5; i++){
        
    
                    //2.构建事件对象
                    SimpleEvent event = new SimpleEvent();
    
                    //3.给事件设置值
                    event.setBody((prefix + "--" + i + subfix).getBytes());
    
                    //4.将事件传给channel
                    getChannelProcessor().processEvent(event);
    
                    status = Status.READY;
                }
            } catch (Exception e) {
        
                e.printStackTrace();
                status = Status.BACKOFF;
            }
    
            try {
        
                Thread.sleep(2000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
    
            return status;
        }
    
        @Override
        public long getBackOffSleepIncrement() {
        
            return 0;
        }
    
        @Override
        public long getMaxBackOffSleepInterval() {
        
            return 0;
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 15]

配置文件

在配置文件中，sink和channel都一样，只是source需要指定自己自定义的，**将代码打包放到flume的lib中**

#声明三个组件
    a1.sources=r1
    a1.sinks=k1
    a1.channels=c1
    
    #声明source的属性
    a1.sources.r1.type=com.gis.source.MySource
    a1.sources.r1.prefix=lnnu
    a1.sources.r1.subfix=done
    
    #声明sink的属性
    a1.sinks.k1.type=logger
    
    #声明channel的属性
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    
    #建立联系
    a1.sources.r1.channels=c1
    a1.sinks.k1.channel=c1

#### 自定义Sink ####

Sink 不断地轮询 Channel 中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。

Sink 是完全事务性的。在从 Channel 批量删除数据之前，每个 Sink 用 Channel 启动一个事务。批量事件一旦成功写出到存储系统或下一个 Flume Agent，Sink 就利用 Channel 提交事务。事务一旦被提交，该 Channel 从自己的内部缓冲区删除事件。

Sink 组件目的地包括 hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、  
自定义。官方提供的 Sink 类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些 Sink。

**自定义Sink步骤**

自定义一个类继承AbstractSink并且实现Configurable接口，实现里面的两个方法

代码编写

package com.gis.sink;
    
    import org.apache.flume.*;
    import org.apache.flume.conf.Configurable;
    import org.apache.flume.sink.AbstractSink;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * @author LnnuUser
     * @create 2021-08-04-上午10:22
     */
    public class MySink extends AbstractSink implements Configurable {
        
    
        //定义两个属性，前后缀
        private String prefix;
        private String subfix;
    
        //获取logger对象
        private Logger logger = LoggerFactory.getLogger(MySink.class);
    
        @Override
        public void configure(Context context) {
        
    
            //读取配置文件，为前后缀赋值
            prefix = context.getString("prefix");
            subfix = context.getString("subfix","lnnu");
        }
    
        /**
         * 1.获取channel
         * 2.从channel获取事务以及数据
         * 3.发送数据
         * @return
         * @throws EventDeliveryException
         */
        @Override
        public Status process() throws EventDeliveryException {
        
    
            //1.定义返回值
            Status status = null;
    
            //2.获取channel
            Channel channel = getChannel();
    
            //3.从channel获取事务
            Transaction transaction = channel.getTransaction();
    
            //4.开启事务
            transaction.begin();
    
            try {
        
                //5.从channel获取数据
                Event event = channel.take();
    
                //6.处理事务
                if (event != null) {
        
    
                    String body = new String(event.getBody());
                    logger.info(prefix + body + subfix);
                }
    
                //7.提交事务
                transaction.commit();
    
                //8.成功提交，修改状态信息
                status = Status.READY;
    
            } catch (ChannelException e) {
        
    
                //9.提交事务失败
                transaction.rollback();
    
                //10.修改状态
                status = Status.BACKOFF;
                e.printStackTrace();
            }finally {
        
    
                //11.最终关闭事务
                transaction.close();
            }
    
            //12.返回状态信息
            return status;
        }
    }

Flume的配置文件

Source和Channel都是正常的写法，只是Sink需要指定为我们自定义的全类名，要先**打包之后上传到flume的lib目录下**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 16]

a1.sources=r1
    a1.sinks=k1
    a1.channels=c1
    
    #声明source的属性
    a1.sources.r1.type=netcat
    a1.sources.r1.bind=localhost
    a1.sources.r1.port=33333
    
    #声明sink的属性
    a1.sinks.k1.type=com.gis.sink.MySink
    a1.sinks.k1.prefix=lnnu
    a1.sinks.k1.subfix=gis
    
    #声明channel的属性
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    
    #建立联系
    a1.sources.r1.channels=c1
    a1.sources=r1
    a1.sinks=k1
    a1.channels=c1
    
    #声明source的属性
    a1.sources.r1.type=netcat
    a1.sources.r1.bind=localhost
    a1.sources.r1.port=33333
    
    #声明sink的属性
    a1.sinks.k1.type=com.gis.sink.MySink
    a1.sinks.k1.prefix=lnnu
    a1.sinks.k1.subfix=gis
    
    #声明channel的属性
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    
    #建立联系
    a1.sources.r1.channels=c1
    a1.sources=r1
    a1.sinks=k1
    a1.channels=c1
    
    #声明source的属性
    a1.sources.r1.type=netcat
    a1.sources.r1.bind=localhost
    a1.sources.r1.port=33333
    
    #声明sink的属性
    a1.sinks.k1.type=com.gis.sink.MySink
    a1.sinks.k1.prefix=lnnu
    a1.sinks.k1.subfix=gis
    
    #声明channel的属性
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    
    #建立联系
    a1.sources.r1.channels=c1
    a1.sinks.k1.channel=c1

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/466d8edef0e642c39a039ba70120e546.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/f5c689772a414d788ee110d6bd10d513.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/7cff3ebaef5b43ee8a42787ce78bf772.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/75a8aa9de6734640a393fc6fdfe89949.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/8c1ee3b0503949f38c15df66e00d0c15.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/45f974f01f2147778ae96f33bf5d14c1.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/079d2072612c4e44a18bc7f3c9bd8aa9.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/3ece0a8b51ea454387ff5c0476085feb.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/fbdecfabd2354100a096cdd9d3c3b6b9.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/073529ffc0ad4584af2714d797674799.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/bac61fd3ea6143de90a9fb1ea3703269.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/b632fba751d34ba5a8e0f5f030ac73b1.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/b1b520b18b0b433a953d603d4b670bb1.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/7dd1bd6e557a4651aa8ba1b01dd41948.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/6176a06ddd254047bb5f95c0ed2a718e.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 15]: https://img-blog.csdnimg.cn/cec2fb6f65bd414ab98d2f5872a40a7a.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 16]: https://img-blog.csdnimg.cn/427add757b29401ab039874dc6da66a1.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70