一、什么是 Flume？

1、Flume 是做什么的？

  Flume 是 Cloudera 开发的一个高可用、高可靠、分布式的海量日志采集、聚合和传输的系统。Flume 支持在日志系统中定义各类数据发送方，用于收集数据。Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力，数据源、数据存储系统都可定制、可扩展。

2、Hadoop 生态圈

  在 Hadoop 生态圈中，数据采集引擎主要是用 Sqoop 和 Flume，其中 Sqoop 主要是面向数据库的数据采集，Flume 主要是对日志的采集。

3、为什么需要 Flume ？

  为什么不直接将日志从生产系统的服务器写入到 HDFS 中，而是需要 Flume 来处理？如果直接写入 HDFS 会产生很多问题：

• 1）HDFS 是主从架构，写入操作会在 NN 上产生一系列复杂的操作，而生产日志又是源源不断产生的，如果产生的日志马上写入到 HDFS，同一时间大量小文件操作或创建会对服务器产生严重的延迟和负担。

• 2）让生产系统直接接触 HDFS 会产生很多管理开销，安全性上也会有很多问题；不同的应用系统可能是用不同的编程语言开发的，运行在不同的环境上，如果直连 HDFS，可能会有各种各样的 bug。

• 3）在多个机房的情况下（混合云，使用不同的云厂商），托管的 Hadoop 集群将数据汇聚到中心节点，需要跨越 WAN 网络，可能会遇到严重的延迟和断线（在互联网上，时延达到20ms已经是非常高质量的网络），需要考虑接入软件对高延迟网络的容忍度和故障恢复能力，会让生产系统产生耦合和开销。

3-mix-cloud.png

• 4）大部分业务的生产流量带有波峰波谷模式（比如早上8、9点流量明显上升，中午吃饭午休下降，下午又迎来一个波峰，下班后下降），为了保证数据不丢失地写入大数据集群，不同业务系统都需要做优化和改动，成本非常高，应该想办法隔离生产应用并保持生产应用数据流的可控。

总结起来，使用 Flume 的理由如下：

• 1）批量处理数据，降低 HDFS NN 处理大量小文件的延迟和负担；

• 2）提高安全性，降低直连管理开销；

• 3）通过 Flume 作为管道系统中间件，将生产系统对高延迟网络的容忍度和故障恢复能力的要求，和多机房环境下 HDFS 可能会因为网络问题导致的严重延迟和断线解耦；

• 4）保持对生产应用数据流的可控，应用系统只需要写入到 Flume，具体要和其他系统日志汇聚，还是沉淀分发到多个处理或存储系统，由 Flume 来处理。

二、Flume 的内部原理

  Flume 被设计成一个灵活的系统，具有以下特点：

• 分布式
• 高度可定制
• 数据可持久化

  分布式，就是说 Flume 可以有多个 agent，每个 agent 就是一条 Flume 服务器，这就保证了有能力来处理海量的数据；高度可定制，指可以采集各种数据源各种格式通过各种协议上传过来的数据，可以选择数据通道是暂存在内存中还是持久化到磁盘中，最后数据沉淀起来，可以选择分发到多个处理系统或者汇聚到某个处理系统；数据可持久化，指采集到 Flume 的数据可以持久化到文件通道中，这样即使 agent 发生故障，恢复后数据不丢失。

Agent

  Flume 中执行具体工作的最小单位叫 agent，一个 agent 可以串到另一个 agent 上，形成一个 agent 链，官方叫多 agent 流。

  Flume Agent 中有三大组件：

• Source：负责获取事件数据到 Flume Agent

• Channel：存储 Source 的数据缓冲区

• Sink：负责从 Agent 取走数据并转发到下一个数据节点，可以是 HDFS HBase 或另一个 Agent

  运行流程如下

三、Flume 通信

  Flume Agent 有各种各样的流模式，最基本的是一对一的简单模式，以及常用的 Fan-out、Fan-in 模式，以及这几种模式互相组合嵌套的复杂流模式

1、Flume 模式

（1）简单模式

（2）Fan-out 模式

（3）Fan-in 模式

（4）复杂流模式

（5）动态路由模式

  动态路由模式的 channel 有一个选择器，能根据请求报文头决定数据流到哪个 sink。

2、失败和数据丢失

  对于多个 agent 组成的工作流，对 agent 故障具有一定的缓冲处理能力。

  如上图所示，由两个 agent 组成的流，假设每个 agent 的容量是 10000 条数据，应用系统每分钟产生 5000 条数据发送给 flume，现在假设终端 sink 到 HDFS 的 agent 到 HDFS 之间的网络发生了故障，因为2个agent能缓冲20000条数据，所以在接下来的4分钟，这个流依然能够正常接收数据知道每个agent都满了，只要能在4分钟内排除故障，那么期间不会有数据的丢失。

  但是假如故障时间超过了4分钟，因为所有agent的容量都满了，无法再缓冲数据，则还是会发生数据的丢失。

3、批量提交模式

  Flume agent 支持批量提交模式，对于大数据场景，批量地处理数据有利于分摊一些公共的开销，从而整体上提高性能和吞吐量。

  Flume 的 source 和 sink 有设置批量大小的参数，该参数确定它们在一个批次中处理的最大 event 数，批量大小必须小于 channel 的交易容量，通常发生在称为事务容量上限的 channel 事务中。

四、环境搭建

  环境配置： ubuntu 16.04 操作系统，jdk-1.8.0_161、apache-flume-1.9.0。

1、ubuntu

  静态化 IP（ /etc/network/interfaces ）

# This file describes the network interfaces available on your system
# and how to activate them. For more information, see interfaces(5).

source /etc/network/interfaces.d/*

# The loopback network interface
auto lo
iface lo inet loopback

# The primary network interface
auto ens33
#iface ens33 inet dhcp

# 使用静态ip的配置
iface ens33 inet static
address 192.168.72.129
netmask 255.255.255.0
gateway 192.168.72.2
dns-nameserver 119.29.29.29

  配置主机名（ /etc/hostname ）

master01

  配置 Host（ /etc/hosts ）

192.168.72.129 master01

  reboot 让配置生效。

2、Java

sudo mkdir /usr/java
sudo tar zxvf jdk-8u161-linux-x64.tar.gz  -C /usr/java/
sudo ln -s /usr/java/jdk1.8.0_161/  /usr/java/default

  配置环境变量（ /etc/profile ）

export JAVA_HOME=/usr/java/default
export CLASSPATH=.:${JAVA_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

  激活 Java 环境

sudo update-alternatives --install /usr/bin/java java /usr/java/default/bin/java 300
sudo update-alternatives --install /usr/bin/javac javac /usr/java/default/bin/javac 300
sudo update-alternatives --config java      
sudo update-alternatives --config javac 

  reboot 重启服务器让配置生效，再用 java -version 命令验证。

3、Flume

  解压安装

sudo mkdir /app 
sudo tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /app
cd /app 
sudo  mv apache-flume-1.9.0-bin/ flume

  配置环境变量（ /etc/profile ）

export FLUME_HOME=/app/flume
export PATH=$PATH:$FLUME_HOME/bin

  让配置生效 source /etc/profile

  验证安装是否成功

flume-ng version

五、三大组件

1、Source

  Source 是负责接受数据到 Flume 的组件，可以从其他系统中接收数据，向 JMS、Kafka、Http、Nc、Tcp、Exec 等，也可以从其他 Flume 代理程序处通过 RPC 管道接收数据，所以 Flume 的数据源接收器可以接收任何来源的数据。

  Flume 自带了很多开箱即用的 Source，通常可以用别名（如avro、nc、syslog）表示。

  启动一个 agent 时，可以指定配置文件，在配置文件中可以根据需要定义需要使用的 source、channel、sink。

  source 的可选参数：

• type：source 的类型，可以是 FQCN 或别名（别名仅为自带 source 使用，如 syslog、avro、netcat 等）。

• channels：数据源写入通道，可设置多个用空格分隔，常用有内存通道和文件通道，如果采用策略路由模式，需要配置特定的选择器 selector。

• interceptors：Flume 的拦截器，用于对相关数据进行高级分类，使用时需要指定相关别名，并配置对应参数。

• selector：Flume 的通道选择器，用于对相关数据进行高级分类，默认使用克隆选择器，按需要可以配置成策略路由选择器。

（1）netcat source

  netcat 是一个基于 TCP 端口的数据源接收器，使用时需要配置如下参数：

• channels
• source type —— netcat
• bind —— 绑定的主机 IP 地址
• port —— 所监听的端口号

现在有一个需求，将 netcat source 接收到的数据打印到标准控制台输出和保存到日志文件中，对于 channel 选择使用内存通道，sink 使用 logger 下沉器，日志输出的格式通过 log4j2.properties 设定，在 log4j2 配置文件中，设定了控制台输出和文件输出，如图所示：

  同时还要求能对网络进行监控

  下面演示一个 netcat source 的使用和配置：

  在 $FLUME_HOME/conf 目录下新增一个 example01.conf 的配置文件

# 配置Agent中的三要素
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 配置Source部分
a1.sources.r1.type = netcat
a1.sources.r1.bind = 192.168.72.129
a1.sources.r1.port = 44444

# 配置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

  a1 表示 agent 的名字；一个 agent 有三要素 source、channel、sink，配置属性后为 s 表示每个要素都可以配置多个，即可以有多个 source、channel、sink，这里设置的名字分别为 r1、c1、k1。

  对于 source 来说，设置类型为 netcat，绑定 agent 机器的IP 192.168.72.129，接收客户端数据的端口为 44444；channel 设置为内存通道，最大容量为 1000，每次最多提交 100 条数据；sink 设置为 logger，即数据会输出到标准输出中，方便查看效果。

  日志输出需要设置 log4j，在 conf 目录下创建一个 log4j.properties，内容如下：

log4j.rootLogger=INFO, A1, A2

# A1这个设置项被配置为控制台输出 ConsoleAppender.
log4j.appender.A1=org.apache.log4j.ConsoleAppender

# A1 输出项的输出格式.
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

# A2 这个配置项的设置，文件输出
log4j.appender.A2=org.apache.log4j.FileAppender

# 设置日志的文件名
log4j.appender.A2.File=/app/flume/logs/log.out

# 定义输出的日志格式
log4j.appender.A2.layout=org.apache.log4j.PatternLayout
log4j.appender.A2.layout.conversionPattern=%m%n

  最后需要把三要素绑定起来，source 才知道数据要保存到哪个 channel，sink 才知道从哪个 channel 获取数据。

  启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/example01.conf --conf /app/flume/conf

  查看 44444 端口，确认进程是否正常启动

  接下来开启一个 telnet 向 192.168.72.129 的 44444 端口发送数据

  通过 wireshark 监控网络流量，可以看到客户端发送给 flume agent 的数据

  同时，在 flume-ng 的进程标准输出中，也看到了客户端发送过来的 11、22、hello world

  查看日志文件

（2）exec source

  exec 是一个基于命令行标准输出来产生数据的数据源接收器，它可以从命令的返回结果中读取数据，将事件转换为 Flume 事件，使用时需要配置如下参数：

• shell —— 具体执行的命令
• source type —— exec，FQCN的全称是org.apache.flume.source.ExecSource
• batchSize —— 每次默认写入channel前累计的最大事件数据量
• charset —— 字符集，默认为UTF-8

现在有一个需求，要对某个日志文件作为数据源，将日志信息输出到控制台和采集到日志中，使用 exec source 来实现。

PS.这样就能够实现对日志的采集，生产系统中会将日志采集到大数据平台进行进一步处理。

  配置文件 example02.conf 内容如下：

# 配置Agent中的三要素
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 配置Source部分
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/flume/logs/access.log

# 配置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

  log4j 配置文件不变，现在启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/example02.conf --conf /app/flume/conf

  若此时 /home/flume/logs/access.log 中有内容，则会输出到标准控制台和日志文件中，往 access.log 中添加内容，也会输出到标准控制台和日志文件。

（3）avro source

  avro 是一个非常重要的 RPC 数据源，被设计成高扩展的 RPC 服务端，可以从其他的 Flume 的 RPC Sink 或者通过 SDK 开发的发送数据客户端接收数据。Avro Source 的可扩展性结合 Channel 担当了缓冲器的角色，使得 Flume 能够处理重要的负载峰值。

  Avro 使用的参数定义如下：

• Source Type：avro，FQCN的全称是org.apache.flume.source.AvroSource
• bind：绑定的IP地址
• port：服务所绑定的端口
• threads：接收客户端数据时最大的工作线程数量

Avro 数据源使用Netty服务器来处理接入请求，Netty服务器使用Java的非阻塞I/O，这保证了当Netty服务器使用相对较少的线程来处理请求的高性能。同时Avro还可以结合Java的JKS进行数据的加密保护。另外对于数据量很大的情况，也可以开启Avro的压缩器，来降低数据传送的容量。

PS. 需要发送接收两端同时开启相同的压缩器才能保证压缩的正常工作。

  现在搭建一个 avro source 的 agent，并编写 Java 程序向该 agent 发送数据。

  实验网络图如下

  配置文件 example03.conf 如下：

# agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 192.168.72.129
a1.sources.r1.port = 44444

# 配置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

  启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/example03.conf --conf /app/flume/conf

  每种 source 能够识别的数据时具有一定的格式的，netcat 只能识别 telnet 发过来的数据，如果telnet 往 avro source 的 agent 发送数据，可以建立连接但是会抛出异常，原因是 avro 识别不了 netcat 格式的数据

  所以要编写 avro 的客户端 Java 程序，引入依赖

<dependency>
    <groupId>org.apache.flume</groupId>
    <artifactId>flume-ng-sdk</artifactId>
    <version>1.9.0</version>
</dependency>

  Java 程序如下

public class AvroDemo {
    public static void main(String[] args) throws Exception {
        String ip = "192.168.72.129";
        int port = 44444;
         
        RpcClient client = RpcClientFactory.getDefaultInstance(ip, port); // Avro
        Event event = EventBuilder.withBody("My password is 123456", Charset.forName("UTF8"));
        client.append(event);   
        client.close();
    }
}

  运行程序，从 agent 的标准输出可以看到发送过去的数据

  查看 Wireshark，可以看到发送的数据

2、Channel

（1）概念

Flume 的数据通道是做什么的？

  Channel 是位于 Source 和 Sink 之间的缓冲区，因此 Channel 允许 Source 和 Sink 运行在不同的速率上，同时 Channel 也是 Flume 保证不丢失数据的关键环节。

  假如没有中间这一层缓冲，那么当 Sink 的处理速度慢于 Source 的接收速度时，超出负荷部分的数据会丢失。

Flume 的数据通道如何链接不同的组件？

  Source 写入数据到一个或多个 Channel 中，再由一个或多个 Sink 将数据取走，Sink 在这一环节中只能从和自己相连的 Channel 读取数据，而多个 Sink 可以从一个相同的 Channel 中读取数据（起到负载均衡的作用），以便获取更好的性能。Channel 提供事务支持，允许 Flume 为写进 Channel 的数据提供明确的保证。

  位于 Source 和 Sink 之间的缓冲区允许它们工作在不同的速率上，因为写操作发生在缓冲区尾部，而读操作发生在缓冲区头部，使得 Flume 能够处理 Source 的高峰负载，即使 Sink 无法立即读取 Channel。

PS. 这里的示意图，通道可以看到是一层一层的，Source 是从下方写入，Sink 是从上面读取。

（2）事务

Flume 的事务特性

  事务就是原子性写入的数据，要么全部批量存在于 Channel 中，要么全部不存在，事务是对数据完整性的一种保护。例如，如果一个 Sink 要将数据存储在文件中，但是在存储到 99% 的时候服务器发生了故障，那么这个事务会重新存储到 Channel 中，这样就可以等服务器恢复后，再次从 Channel 中获取数据进行重写。

  上图是 Flume 事务处理的流程图，Put 是写数据，Take 是读数据。

  当发生数据的读写时：

• 每次都会判断是否发生错误或失败，如果是则事务回滚，整个流程结束。

• 若正常读写，则判断事件是否为空，是则提交事务；
  若否则表示正常读写，判断是否达到事务批次数量，没有则继续读写，达到限制则提交事务

• 提交事务时，失败报错则回滚，最终流程结束。

（3）种类

  channel 的种类很多，常用的有 Memory Channel（内存通道）、File Channel（文件通道），内存通道性能非常突出，是常用的通道；如果对数据可靠性有要求，即使服务器故障断电也不丢失数据，就可以使用文件通道，但是性能跟内存通道的差了几个数量级。

  除此之外，还有将数据保存到数据库的 JDBC Channel；将数据保存到 MQ 的 Kafka Channel；Spillable Memory Channel 是可溢出内存通道，当缓存数据超出内存容量设置时，将输入数据时间溢出保存到内嵌的文件通道中；Pseudo Transaction Channel 仅用于单元测试，不用于生产；如果想自定义通道，可以实现 Flume 提供的 Custom Channel 接口。

① 内存通道

  内存通道就是建立在内存中的通道，在 JVM 的堆上存储写入的数据事件，事实上内存通道的队列是 Source 从它尾部写入，Sink 从它头部读取。

  由于内存的带宽非常高，例如一条 DDR 4 3200MHz 的内存，它每秒的吞吐速度可以达到 25GB，所以使用内存通道的 Flume 代理有着极高的运行速度。（3200x8=25600）

  内存通道的缺点是不能支持意外断电事故，如果出现断电，内存里的数据会全部丢失，所以内存通道的配置可以用在允许数据出现微量丢失的情况，加入需要完全不丢失数据，可以用文件通道模式。

事务特性

  内存通道支持事务特性，一旦一个 Source 事务提交，该事务队列被自动移入内存通道的主队列，并对 Sink 通道可见。如果失败了，Source 将回滚，主队列中不会出现相关数据。相对的，Sink 部分在读取数据后需要保证是否已经在后端存储完成数据数据的保存，只有正确提交后，才可以向队列发送事务结束，这样数据就会从主队列中标记为可以废弃，等待垃圾回收。

参数设置

• type：memory（内存通道类型）。

• capacity：channel 可以保存的提交事务的最大容量值。

  每台服务器上的内存数量都是有限的，例如 8GB 16GB 等，所以这个容量是指，在任何时候都能保持的被提交的事务的总和，是 Source 存入值 Sink 取出值的差值。如果存储的事务达到了设置的最大值，那么 channel 会抛出 ChannelException，直到事件别取走，恢复正常。

• ByteCapacity：该参数限定 channel 中事件可以使用的内存总量。

• transactionCapacity：单次事件可以写入或读取的最大事件数量。

  这个参数的一个妙用是可以通过设置偏小的阈值，防止流氓客户端一下子推送大量事件，导致 Flume 程序瞬间过载，程序崩溃。这个参数强制指定了批量的大小，从而限制了每次调用可提交的最大数，是对付 DOS 攻击的最好的一个方法。

  使用容量相关参数时，要注意 flume agent 的 Java 进程内存大小设置，默认情况下最大只会分配 20M 内存

  需要调整启动的 vm 参数，修改 $FLUME_HOME/conf/flume-env.sh 配置

...
export JAVA_OPTS="-Xms1000m -Xmx3000m -Dcom.sun.management.jmxremote"
...

【实战】现在要配置这样一个内存通道：可以保存10万个事件，每个事务可以保存1000个事件，channel中可以占用的内存是5GB，在5GB中，事件头header可以占500MB，主题事件占4.5GB。

...
agent.channels = mc

agent.channels.mc.type = memory
agent.channels.mc.capacity = 100000
agent.channels.mc.transactionCapacity = 1000
agent.channels.mc.byteCapacity = 5000000000
agent.channels.mc.byteCapacityBufferPercentage = 10

...

② 文件通道

  文件通道会将数据持久化到磁盘文件中，与内存通道动辄几十GB每秒的读写速度相比，文件通道的读写速度达到100MB/s就已经是很快的速度了，相差几百倍，因此，除非有某些应用场景对数据的可靠性有比较高的要求，否则一般都使用内存通道。

（4）实战

  定义一个配置文件 channel-demo.conf，内容如下：

# 配置Agent中的三要素
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 配置Source部分
a1.sources.r1.type = netcat
a1.sources.r1.bind = 192.168.72.129
a1.sources.r1.port = 44444

# 配置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

  数据源使用的是 netcat，使用内存通道，容量是1000个事件，每次事务批量最多100个，sink 使用 logger。

  启动 Flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/channel-demo.conf --conf /app/flume/conf

  使用 telnet 192.168.72.129 44444 连接到 agent 并发送数据，可以明显观察到，数据回车发送之后过了几秒，agent 的标准输出才打印了事件信息，这是因为数据没有达到事务批次容量 100，所以会等几秒再批量读取数据，这是通过 channel 的 keep-alive 参数（默认是3秒超时）来设定的。

  有时候我们可能想查看监控，观察通道内部堆积了多少事件，事件读写的成功率怎么样，就是需要开启监控，在上述命令基础上添加启动参数：

flume-ng agent --name a1 --conf-file /app/flume/conf/channel-demo.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  这里开放了 5653 端口，可通过 HTTP 访问实时获取通道信息。

  这里我使用 postman 工具访问：http://192.168.72.129:5653/metrics，可以看到以下监控指标信息

各项指标的含义如下：

• ChannelFillPercentage：堆积在通道中的事件占通道容量的百分比。

• ChannelCapacity：通道容量。

• Type：通道类型。

• ChannelSize：目前在通道中的事件数量。

• EventTakeSuccessCount：从通道中成功拿走事件的次数。

• EventTakeAttemptCount：尝试从通道中拿走事件的次数，agent 有后台进程定期 attempt，因此即使没有发数据，该指标也会不断增长。

  <img src="Flume\drawing\39-channel-metric.png" style="zoom:70%;" />

• StartTime：通道启动时间。

• EventPutAttemptCount：向通道尝试写入数据的次数。

• EventPutSuccessCount：向通道中成功写入数据的次数。

• StopTime：通道停止时间。

  发送9条数据，可以看到监控指标的变化

  假如发送到 channel 中的数据没有取出，就会堆积在通道中，设置一个有问题的 sink，来看下这个效果，定义配置文件 channel-demo2.conf ，内容如下：

# 配置Agent中的三要素
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 配置Source部分
a1.sources.r1.type = netcat
a1.sources.r1.bind = 192.168.72.129
a1.sources.r1.port = 44444

# 配置Sink部分
a1.sinks.k1.type = avro
a1.sinks.k1.hostname =192.168.72.129
a1.sinks.k1.port = 44443

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 1000

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  启动会抛异常，但可以正常向该 agent 发送数据

  发送一些数据时候，发现 ChannelSize 跟发送的数据条数一致，因为 Sink 无法从通道中取走数据。

  假如通道已满，则后续发送的数据会被拒绝，并报错容量已满，将channel-demo2.conf 中的容量设置为一个较小的值，方便展现这种情况：

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10
a1.channels.c1.transactionCapacity = 10

  发送前10个数据都正常，从监控指标看到通道已满

  再发送第11条数据时，报错数据提交到队列中获取空间失败

3、Sink

（1）概念

Flume 的 Sink 是做什么的？

  从 Flume Agent 移除数据并写入到另一个 agent 或数据存储或一些其他系统的组件被称为 Sink。Flume 还提供了接口，允许用户自定义 Sink，使用时允许用户配置内置的 Sink 或者自定义 Sink。

Sink 的事务特性

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

Sink 的生命周期

  每个被正确配置的 Sink 启动后都会归属于一个 Sink 组，每个 Sink 组包含一个或多个 Sink，如果一个 Sink 没有定义 Sink 组，那么系统自动产生一个 Sink 组，且该 Sink 是这个组的唯一成员。如果在一个 Sink 组内有多个 Sink，则通过循环调用的方式依次处理组中每一个 Sink 的 Process 方法。

PS.循环调用就是串行方式处理，每个时间点只有组内的一个 Sink 处理数据，并不能提高数据处理的速度。

  如何通过 Sink 提高性能？

  通常一个线程运行一个 Sink，如果是大规模的 IO 密集操作，可能会出现来不及处理数据的情况。Sink 会来不及 及时将数据提交，并移除出 Channel，导致后续数据堆积甚至堵塞（Channel 堆积数据容量是有参数控制的，不是无限的），当达到 Channel 容量限制时就会拒绝后续提交过来的数据。

  可以通过添加多个 Sink 到不同的 Sink 组中从不同的 Channel 来获取数据，这对于 HDFS、HBase 等大数据集群来讲尤为有用，可以大幅提高谢瑞数据和处理性能。

  比如现在有100G数据要通过 Flume 上传到 HDFS，假设我们有50台机器，每条机器上都有单独的 sink 和 channel，那就可以并发处理，平均下来每台机器处理2G的数据，总的处理时间可以减小到单台机器处理的1/50。

（2）类型

  跟 Channel 一样，Flume 提供了丰富的 Sink，有用于大数据的 HDFS Sink、HBase Sink、HIve Sink，用于日志的 Logger Sink，常用于 Flume 链之间通信的 Avro Sink，用于索引系统的 Solr Sink、Elastic Search Sink，支持自定义 sink 的 Custom Sink等。

（3）HDFS Sink

为什么使用 HDFS Sink？

  当 HDFS 客户端直接提交数据到 HDFS 时，会产生很多小文件，给 HDFS 的 NameNode 造成很大的压力，因此建议使用 Flume Agent 来写入数据。

工作机制

  HDFS 的 bucket（桶）可以理解为是一个目录，在 HDFS Sink 中，可以同时将数据写入多个 bucket 中，不过每个事件将只能进入到一个 bucket 中，也就是任何一个时间点，每个 bucket 只能打开一个文件，多个 Sink 可以在不同的 bucket 打开不同的目录文件。Flume 写入数据后，HDFS 会出现以 .tmp 扩展名结尾的文件，等文件写完关闭后，自动消除 .tmp 后缀。

文件滚动策略

① 基于时间的策略

配置项：hdfs.rollInterval
默认值：30秒

说明：如果设置为0表示禁用这个策略

原理：在 org.apache.flume.sink.hdfs.BucketWriter.append 方法中打开一个文件，都会调用 open 方法，如果设置了 hdfs.rollInterval，那么 hdfs.rollInterval 秒之内只要其他策略没有关闭文件，文件就会在 hdfs.rollInterval 秒之后关闭。

② 基于文件大小和事件数量策略

配置项：
文件大小策略 - hdfs.rollSize（默认值1024字节）
事件数量策略 - hdfs.rollCount（默认值10）
说明：如果设置为0表示禁用这个策略

③ 基于文件闲置时间策略

配置项：hdfs.idleTimeout（默认值0）

原理：如果文件在 hdfs.idleTimeout 内都是空闲的，没有任何数据写入，那么当前文件关闭，滚动到下一个文件。

【配置案例】设置一个 agent 配置，要求将事件写到10分钟的bucket中，使用基于时间的分桶，采用Snappy压缩器，每2分钟或者10万个事件或者128M写到一个文件，无论哪个条件先达到，Sink 都执行文件关闭，如果文件打开30秒没有写入，也关闭此文件，最多打开100个文件，超过100个同时打开的文件，也关闭相应文件。

agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsChannel.channel = memory
agent.sinks.hdfsSink.hdfs.path = /Data/UsingFlume/%{topic}/%Y/%m/%d/%H/%M
agent.sinks.hdfsSink.hdfs.filePrefix = UsingFlumeData
agent.sinks.hdfsSink.hdfs.inUsePrefix = .
agent.sinks.hdfsSink.hdfs.inUseSuffix = .temporary
agent.sinks.hdfsSink.hdfs.fileType = CompressedStream
agent.sinks.hdfsSink.hdfs.codeC = snappy
agent.sinks.hdfsSink.hdfs.rollSize = 128000000
agent.sinks.hdfsSink.hdfs.rollCount = 100000
agent.sinks.hdfsSink.hdfs.rollInterval = 120
agent.sinks.hdfsSink.hdfs.idleTimeout = 30
agent.sinks.hdfsSink.hdfs.maxOpenFiles = 100
agent.sinks.hdfsSink.hdfs.round = true
agent.sinks.hdfsSink.hdfs.roundUnit = minute
agent.sinks.hdfsSink.hdfs.roundValue = 10

（4）实战

  首先在 flume agent 本地启动 hdfs 服务，创建 hdfs-demo.conf，内容如下：

# agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# 配置Sink部分
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/flume/logs

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  启动 fluent agent

flume-ng agent --name a1 --conf-file /app/flume/conf/hdfs-demo.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  我们往 agent 的 44444 端口通过 avro 连接（AvroDemo.java）发送数据，发送数据后发现，原来 HDFS 中不存在的 /user/flume/logs 路径被自动创建了，并且在该目录下创建了文件，打开 web console（http://192.168.72.129:50070/）查看。

  默认情况下，每隔30秒文件会滚动一次，滚动后 .tmp 文件会去掉后缀，如上图所示。

  如果再次发送数据，则会生成新的临时文件。

  不断发送数据，则在30秒的周期内，发送的数据都会被保存到该临时文件中，文件会不断变大，当30秒周期结束后，临时文件再次滚动为 FlumeData 文件。

  从上面的实验过程，可以看到基于时间策略 hdfs.rollInterval 参数作用的效果。

  查看文件内容，可以看到发送的数据为 "My password is 123456"，文件是保存二进制数据的顺序文件。

  下面的实验演示时间数量策略，为了避免其他策略的影响，将滚动周期调整为1小时，文件大小滚动阈值设置为0（禁用），hdfs-demo2.conf 配置如下所示：

# agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# 配置Sink部分
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/flume/logs/data_%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = retail
a1.sinks.k1.hdfs.fileSuffix = .txt
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize  = 0
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.useLocalTimeStamp = true

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  sink 配置中，会根据系统时间年月日来创建文件夹，每小时创建一个文件夹，每处理五个时间即滚动文件，文件前缀是 retail，后缀是 .txt，文件类型是数据流，使用本地时间戳。

  启动 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/hdfs-demo2.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  在还没有写入数据时，不会创建目录，现在发送一些数据，内容为 "My password is 123456"，一共发送五条，查看监控指标，可以看到 sink 成功 “流走” 了5条数据，即将这五条数据存储到了 hdfs 中。

  查看 hdfs 可以看到生成了目录和临时文件

  当发送第6条数据时，触发基于事件数的文件滚动策略，临时文件滚动，并生成新的临时文件存储第6个事件数据，为了区分第6次发送保存到了哪个文件，将发送内容调整为 "My password is 654321"。

  查看文件内容，证明第六次发送的数据确实保存到了新的临时文件中。

六、高级特性

1、拦截器

（1）概念

  拦截器（interceptor）是一个简单插件组件，工作在 Source 和 Channel 之间，在 Source 接收到数据之后，拦截器可以基于任何自定义标准删除或转换相关事件。如果在一个链路中有多个拦截器，那么拦截器将按设定的顺序依次执行。

（2）类型

  拦截器的种类很多，Flume 内置了很多拦截器，所有拦截器都要实现 org.apache.flume.interceptor.Interceptor 接口，常用的拦截器有时间戳拦截器、主机拦截器、静态拦截器、正则拦截器等。

  使用拦截器时要指定类型，对于内置的拦截器可以使用别名，对于自定义拦截器必须指定 FQCN。

（3）实战

① 时间戳拦截器

  时间戳拦截器会在 event 的 headers 中添加时间戳，interceptor-time.conf 示例配置：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44443
# +++++++++++新增如下配置 检查拦截器是否工作
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp

# 配置Sink部分
a1.sinks.k1.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/interceptor-time.conf --conf /app/flume/conf

  发送数据，可以在日志中看到 headers 中加上了一个时间戳的字段

② 主机拦截器

  主机拦截器会将当前 agent 的 IP 或 hostname 添加到 headers 中，默认情况下是添加 IP，示例配置（interceptor-host.conf）：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44445
# ++++++++++++++++ 新增主机拦截器 ++++++++++++++
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = host
# 输出主机名而不是IP
#a1.sources.r1.interceptors.i1.preserveExisting = true
#a1.sources.r1.interceptors.i1.useIP = false
#a1.sources.r1.interceptors.i1.hostHeader = hostname

# 配置Sink部分
a1.sinks.k1.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/interceptor-host.conf --conf /app/flume/conf

  发送数据，可以在日志中看到 headers 中加上了一个时间戳的字段

  如果想加的是 hostname，则将注释掉的三行配置生效

③ 静态拦截器

  静态拦截器可以向Event header中写入一个固定的键值对属性。不支持写入多个属性，但是可以通过配置多个静态属性写入拦截器来实现。

  示例配置：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44446

a1.sources.r1.interceptors = i1 i2
# ++++++++++++++++ 新增主机拦截器 ++++++++++++++
a1.sources.r1.interceptors.i1.type = host
# 输出主机名而不是IP
a1.sources.r1.interceptors.i1.preserveExisting = true
a1.sources.r1.interceptors.i1.useIP = false
a1.sources.r1.interceptors.i1.hostHeader = hostname

# ++++++++++++++++ 新增静态过滤器 ++++++++++++++
# 这是个很实用的功能，比如标识数据是通过哪台 agent 处理的
a1.sources.r1.interceptors.i2.type = static
a1.sources.r1.interceptors.i2.key = datacenter
a1.sources.r1.interceptors.i2.value = NEW_YORK

# 配置Sink部分
a1.sinks.k1.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  启动 agent，发送数据

  可以看到配置多个拦截器时，会按照配置的顺序先让主机拦截器起作用，再让静态拦截器起作用。

④ 自定义头部 K-V

  经常有一些需求，需要往 event headers 中加入自定义的 K-V，示例代码如下：

String ip = "192.168.72.129";
int port = 44446;   

RpcClient client = RpcClientFactory.getDefaultInstance(ip, port); // Avro

Map<String, String> headers = new HashMap<String, String>(); 
headers.put("ClientServer", "Client01srv");

Event event = EventBuilder.withBody("info", Charset.forName("UTF8"), headers);
client.append(event);       
client.close();

⑤ 正则过滤拦截器

  有时候，只想让消息体中带某些标识的 event 通过过滤，可以使用正则拦截过滤器，配置示例如下：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44447

a1.sources.r1.interceptors = i1

# ++++++++++++++++ 新增正则过滤器 ++++++++++++++
# 只允许消息体包含 "info" 的 event 通过，不符合过滤器的事件并不会在通道中滞留
a1.sources.r1.interceptors.i1.type = regex_filter
a1.sources.r1.interceptors.i1.regex = .*info*
a1.sources.r1.interceptors.i1.excludeEvents = false

# 配置Sink部分
a1.sinks.k1.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  测试代码如下：

String ip = "192.168.72.129";
int port = 44447;      

RpcClient client = RpcClientFactory.getDefaultInstance(ip, port); // Avro

Map<String, String> headers = new HashMap<String, String>(); 
headers.put("ClientServer", "Client01srv");

Event event = EventBuilder.withBody("info", Charset.forName("UTF8"), headers);
Event event2 = EventBuilder.withBody("debug", Charset.forName("UTF8"), headers);

client.append(event);       
client.append(event2);

client.close();

  发送的两条数据，只有 info 会进入通道处理。

2、Channel 选择器

（1）概念

  通道选择器是决定 Source 接收一个事件后，到底要写入哪个通道的组件。通道选择器是通过 Channel 处理器完成的；在 Source 组件配置时用 selector 进行配置，Flume 内置的通道选择器有两种：复制模式、多路复用模式。

复制 Channel 选择器

  如果没有指定通道选择器，则 Source 使用复制 Channel 选择器，该模式会复制每个事件到相关的 Channel 中，相当于流量克隆，每个通道都会拿到一样的数据。由于有多个通道的存在，如果某一个通道故障，则整个事务会报错，此时可以使用 option 选项，收工指定哪些通道是必须成功的，哪些通道是允许失败的。

【示例】假设现在要把数据发送到c1、c2、c3三个通道，其中c3如果出现异常，则不会报错，配置如下：

agent.sources.avroSource.type = avro
agent.sources.avroSource.channels = c1 c2 c3
agent.sources.avroSource.selector.optional = c3

多路复用 Channel 选择器

  多路复用选择器是一种专门用于动态路由事件的选择器，通过选择事件应该写入的 Channel，基于一个特定的事件头的值进行路由，结合拦截器，可以在事件上做一些分析工作，然后决定应该写入哪个 Channel。多路 Channel 选择器会寻找报文头中特定的值，如果未指定特定事件值，则该事件进入默认 Channel。

【示例】配置一个多路复用通道选择器，通道一共有c1、c2、c3、c4、c5，根据事件头中的 priority 字段决定要路由到哪个通道，如果 priority = 1 则路由到 c1、c2，c3可选；如果 priority = 2 路由到 c2，c4可选；其他情况默认路由到c5，并且对 priority = 3 的情况 c4 可选。对应的配置如下：

agent.sources.avroSource.type = avro
agent.sources.avroSource.channels = c1 c2 c3 c4 c5
agent.sources.avroSource.selector.type = multiplexing
agent.sources.avroSource.selector.header = priority
agent.sources.avroSource.selector.mapping.1 = c1 c2
agent.sources.avroSource.selector.mapping.2 = c2
agent.sources.avroSource.selector.optional.1 = c3
agent.sources.avroSource.selector.optional.2 = c4
agent.sources.avroSource.selector.optional.3 = c4
agent.sources.avroSource.selector.default = c5

（2）实战

  复制通道选择器配置范例如下：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# 配置Sink部分
a1.sinks.k1.type = logger
a1.sinks.k2.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

a1.channels.c2.type = memory
a1.channels.c2.capacity = 100
a1.channels.c2.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

  启动 agent 后可以看到有两个 sink 和两个 channel

  发送两条数据，可以看到有4个日志输出，因为有2个channel

  查看监控指标，可以看到每个通道都处理了两条数据

  多路复用通道选择器，范例配置如下：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# 添加选择器 
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = op
a1.sources.r1.selector.mapping.1 = c1
a1.sources.r1.selector.mapping.2 = c2
a1.sources.r1.selector.default = c2

# 配置Sink部分
a1.sinks.k1.type = logger
a1.sinks.k2.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

a1.channels.c2.type = memory
a1.channels.c2.capacity = 100
a1.channels.c2.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

  启动 agent，不带 headers op 属性发送20条数据，测试代码如下：

public class SelectorTest {

    public static void main(String[] args) throws Exception {
        String ip = "192.168.72.129";
        int port = 44444;       
        
        RpcClient client = RpcClientFactory.getDefaultInstance(ip, port); // Avro       
        Map<String, String> headers = new HashMap<String, String>(); 
        //headers.put("op", "1");
        // 批次发送
        List<Event> events = new ArrayList<Event>();
        for (int i = 0; i < 20; i++) {
          events.add(EventBuilder.withBody("info: " + i, Charset.forName("UTF8"), headers));
        }
        client.appendBatch(events);     
        client.close();
    }
}

  查看监控指标，因为没有指定 op，所以默认是发送到 c2

  设置 op 为1，再次发送20条数据

  c1 的事件处理指标变成了20

3、Sink 组

（1）概念

  Flume 中多个 Sink 可以添加到同一个 Sink 组中，Sink 会有一个对应的 Sink Processor，Sink 运行器会要求 Sink 组中的一个 Sink 从自己的 Channel 中读取数据，特别注意 Sink 组中的多个 Sink 是不能同时工作的，每次只能是其中的一个。

  Sink 组通常用在 RPC Sink，主要用来解决层与层之间的负载均衡或故障转移备份用。RPC Sink 被设计成只能有一个 RPC Source，即 Sink 不能发送给多个链路下的多个 agent，所以多层之前的 Flume，需要有相等数量的 Source 和 Sink。

负载均衡

  负载均衡有利于提升处理性能，在 Flume 链中，前面的 agent 只是对请求数据的简单处理缓存分流，处理速度比较快；而后面的 agent 离存储系统比较近，需要等待存储系统将数据持久化，速度比较慢。为了平衡链路前后数据处理读取不一致的问题，需要在链路的末端加多一些 agent 做负载均衡，同时也起到一个缓冲的作用。

【示例】配置一个负载均衡的 sink 组，组中一共有 s1、s2、s3、s4 四个 sink，通过随机方式分配负载，sink 异常的退避时间为 10 秒，如果发生异常，成倍增加退避时间。

agent.sinks = s1 s2 s3 s4
agent.sinkgroups = sg1
agent.sinkgroups.sg1.sinks = s1 s2 s3 s4

故障转移

  故障转移模式下，就是冗余备份，一些 sink 只有在另一些 sink 发生故障的时候才会开始工作。故障转移组逻辑处理器维护了一个发送Event失败的sink的列表，保证有一个sink是可用的来发送Event。

  故障转移机制的工作原理是将故障sink降级到一个池中，在池中为它们分配冷却期（超时时间），在重试之前随顺序故障而增加。 Sink成功发送事件后，它将恢复到实时池。sink具有与之相关的优先级，数值越大，优先级越高。 如果在发送Event时Sink发生故障，会继续尝试下一个具有最高优先级的sink。 例如，在优先级为80的sink之前激活优先级为100的sink。如果未指定优先级，则根据配置中的顺序来选取。

  要使用故障转移选择器，不仅要设置sink组的选择器为failover，还有为每一个sink设置一个唯一的优先级数值。 可以使用 maxpenalty 属性设置故障转移时间的上限（毫秒）。

【示例】配置一个 sink 组，组内有 s1、s2、s3、s4 四个 sink，处理器类型为故障转移，s1、s2、s4 的优先级分别是90、100、110，故障转移上限时间为10秒。

agent.sinkgroups = sg1
agent.sinkgroups.sg1.sinks = s1 s2 s3 s4
agent.sinkgroups.sg1.processor.type = failover
agent.sinkgroups.sg1.processor.priority.s1 = 90
agent.sinkgroups.sg1.processor.priority.s2 = 100
agent.sinkgroups.sg1.processor.priority.s3 = 110
agent.sinkgroups.sg1.processor.maxpenalty = 10000

（2）实战

  负载均衡配置范例：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# 配置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.sinks.k1.type = logger
a1.sinks.k2.type = logger

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

  处理器选择器指定了轮训算法，即对 k1、k2 轮流来处理 channel 事件，发送 10 条数据，每个 sink 各处理了五条，从日志中可以看到使用了 LoadBalancingSinkProcessor。

  故障转移配置范例：

# 配置Agent中的三要素
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1

# 配置Source部分
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44445

# 配置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.sinks.k1.type = logger
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = localhost
a1.sinks.k2.port = 44444

# 配置Channel部分
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

# 绑定相关组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

  sink 组中，配置了 k2 类型为 avro，端口为44444，因为 k2 是一个有问题的 sink（本地并没有程序监听处理44444端口发送的事件），因为会触发故障转移，发送给该 agent 的事件都会由 k1 来处理。

七、项目案例实战

1、项目需求

  某电商项目预计上线一个购物平台，现在需要通过平台获取用户的浏览网站信息和下单信息，要求架构师设计整个后台系统架构，完成如下几项要求：

• 1、完成前端用户行为日志数据采集
• 2、完成数据的后端存储，满足业务部门 HBase 数据库的实时查询和数据分析部门 HDFS 数据存储需求
• 3、要求整个系统的网络不可中断，出现故障时自行故障转移
• 4、所有数据在中间传送过程中要求加密
• 5、所有的数据接入环节都需要有监控，并接入监控平台，方便统一调度

2、需求分析

  对需求进行分解，分解的维度按照以下几个维度展开：

• 业务需求
• 性能需求和灾备需求
• 安全需求
• 管控需求

  业务需求是最基本的需求，即能满足需要的功能，不考虑其他的，可以看成是一个原型，如下图所示：

【分析】

  只需要一个 Flume agent，数据采集和代理部署在同一台机器上，再分别 sink 到 HDFS 和 HBase，但是从图中可见，链路中任何一个地方发生故障，都会导致数据采集不可用，非常脆弱，并且一旦发送过来的数据量比较大，可能会发生数据将 agent 中的内存堆积满拒绝新的数据进入。

  为了满足性能需求和灾备需求，模型演化为以下的拓扑结构

【分析】

  将 Flume 的链路划分为3个层次，首先是直接接收应用系统数据的数据采集层，这里可以部署比较多的 flume agent，采集到数据后汇聚到核心层，核心层会将数据路由到不同存储系统的写入层，写入层起到一个写入前缓冲的作用，最后再持久化到对应的存储系统。

  在每个层，都是由多台 flume agent 组成的，允许故障转移，所以不必担心某台 agent 故障导致数据采集失败，同时也满足性能上的需求。

  数据在网络上直接传输是不安全的，可以被抓包软件获取到数据包的内容，因此基于安全需求，必须对数据进行加密，使用 SSL 加密数据流，保证流量安全。

  最后考虑管控需求，在生产环境中，动辄几十上百台服务器，要实时了解每台服务器的健康状况，需要部署监控，因此还需要有一个 Flume 监控服务器。

3、实战

  为了完成上述的需求，这里使用 6 台服务器

hostname	IP	功能
flume131	192.168.72.131	数据采集Flume代理
flume132	192.168.72.132	核心层路由Flume
flume133	192.168.72.133	HDFS写入层Flume
flume134	192.168.72.134	HBase写入层Flume
flume135	192.168.72.135	HBase数据库
master01	192.168.72.129	HDFS
flume136	192.168.72.136	Ganglia

  整体需求设计图如下：

（1）数据采集器

  先搞定数据采集器，配置（srv01.conf）如下：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
# 测试
a1.sinks.k1.type = logger

# sink 到核心层Flume
#a1.sinks.k1.type = avro
#a1.sinks.k1.hostname = flume132
#a1.sinks.k1.port = 44444

# Use a channel which buffe\rs 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

  在 srv01 上启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/srv01.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  发送数据，srv01 日志打印接收事件正常

（2）核心路由

  核心路由Flume，配置（srv02.conf）如下：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
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

  在 srv02 上启动 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/srv02.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  修改 srv01.conf 的配置，sink 改成到 flume132

# Describe the sink
# 测试
#a1.sinks.k1.type = logger

# sink 到核心层Flume
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = flume132
a1.sinks.k1.port = 44444

  重启 srv01 agent，启动成功后，可以看到 srv02 打印了以下日志，接受了来自 srv01 的连接

  向 srv01 发送数据，可以看到 srv02 日志打印了发送过来的数据

（3）HDFS写入层

  srv03 专心接受来自路由层的要写入HDFS的数据，配置（hdfs-flume.conf）如下：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = master01
a1.sinks.k1.port = 44444

# 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

  在 master01 上启动 HDFS，并配置接受写入层数据到 HDFS 的 agent，配置（master01.conf）如下：

a1.sources = r1
a1.sinks = k1
a1.channels =c1

a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/logs2

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

  在 master01 上先启动 master01 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/master01.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  再在 srv03 上启动 srv03 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/hdfs-flume.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  启动完后可以看到 master01 接受了来自 srv03 的连接

  修改 srv02.conf，配置如下：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = flume133
a1.sinks.k1.port = 44444

# 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

  启动之后，可以看到 srv03 接收到来自 srv02 的连接

  同样需要重启 srv01，可以看到 srv02 接收到来自 srv01 的连接

  至此，整体链路上所有的 agent 都已经准备就绪。现在尝试向 srv01 发送数据，通过 postman 查看 master01 监控指标，可以看到接收到了一条数据

  查看 HDFS web console，可以看到创建了 /user/logs2 目录，并在该目录下创建了 FlumeData 文件

  至此成功实现了经过 Flume 数据写入到 HDFS 的过程。(＾－＾)V

（4）HBase 写入层

  srv04 接收来自 Flume 的要发送给 HBase 的数据，配置如下（hbase-flume.conf）：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = flume135
a1.sinks.k1.port = 44444

# 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

  在 srv05 上安装了 HBase，在配置文件 hbase-site.xml 中配置了将数据存储在 HDFS 上

<configuration>
    <!--数据存储在HDFS上-->
    <property>
        <name>hbase.rootdir</name>
        <value>hdfs://192.168.72.129:9000/hbase</value>
    </property>     

    <!--表示是一个分布式的环境-->
    <property>
        <name>hbase.cluster.distributed</name>
        <value>true</value>
    </property> 

    <!--ZK的地址-->
    <property>
        <name>hbase.zookeeper.quorum</name>
        <value>192.168.72.135</value>
    </property>

    <!--Region的冗余-->
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>

  flume agent 配置（srv05.conf）如下：

a1.sources = r1
a1.sinks = k1
a1.channels =c1

a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# 写入的HBase的表test，列族CF1中
a1.sinks.k1.type=hbase
a1.sinks.k1.table=test
a1.sinks.k1.columnFamily=CF1
a1.sinks.k1.serializer=org.apache.flume.sink.hbase.SimpleHbaseEventSerializer

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

  首先启动 master01 上的 HDFS，然后启动 srv05 上的 hbase

start-hbase.sh

  使用HBase shell 创建表和列族

hbase(main):001:0> create 'test','CF1'
0 row(s) in 2.6680 seconds

  接着启动 srv05 上的 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/srv05.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  启动 srv04 上的 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/hbase-flume.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  将 srv02.conf 下沉的 agent 由 flume133 改成 flume134

...
a1.sinks.k1.hostname = flume134
...

  重启 srv02

flume-ng agent --name a1 --conf-file /app/flume/conf/srv02.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  重启srv01

flume-ng agent --name a1 --conf-file /app/flume/conf/srv01.conf --conf /app/flume/conf -Dflume.monitoring.type=http -Dflume.monitoring.port=5653

  至此已经将所有 agent 和环境都搞定了，用以下代码发送数据：

@Test
public void testHBaseFlume() throws EventDeliveryException {
    String ip = "192.168.72.131";
    int port = 44444;
    RpcClient client = RpcClientFactory.getDefaultInstance(ip, port);
     
    List<Event> list = new ArrayList<Event>();
    for (int i = 1; i <= 10; i++) {
        Event event = EventBuilder.withBody("test srv01 -> srv02 -> hdfs", Charset.forName("UTF8"));
        list.add(event);
    }
    client.appendBatch(list);
    client.close();
}

  查看 srv05 的监控指标，可以看到10条数据进来了

  查看 hbase 的表数据，可以看到写入了10行

（5）SSL 数据加密

  srv01 模拟了生产的 flume 客户端，srv01 到 srv02 之间的网络是互联网，如果有人在网络上使用抓包工具，是可以看到 srv01 发送给 srv02 的明文的。

  现在测试发送 "test info" 的明文，可以抓包到明文。

  在互联网上明文传输数据时非常不安全的，因此需要使用 SSL 进行加密。

  首先，生成密钥库文件 keystore.jks，并且放到 srv01、srv02 的 /app/flume/conf 目录下，SSL使用非对称密钥来对通道进行加密

keytool -genkey -keyalg RSA -alias selfsigned -keystore keystore.jks -storepass password -validity 360 -keysize 2048

  修改 srv01.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = flume132
a1.sinks.k1.port = 44444
a1.sinks.k1.ssl = true
a1.sinks.k1.truststore = /app/flume/conf/keystore.jks
a1.sinks.k1.truststore-password = password
a1.sinks.k1.truststore-type = jks

# Use a channel which buffe\rs 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

  修改 srv01.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
a1.sources.r1.ssl = true
a1.sources.r1.keystore = /app/flume/conf/keystore.jks
a1.sources.r1.keystore-password = password
a1.sources.r1.keystore-type = jks

# 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

  再次先后启动 srv02、srv0，再次抓包到的数据已经被加密

（6）Ganglia 监控

  监控使用 ganglia，首先配置 host，在 flume136 的 /etc/hosts 中添加配置：

192.168.72.129 master01
192.168.72.131 flume131
192.168.72.132 flume132
192.168.72.133 flume133
192.168.72.134 flume134
192.168.72.135 flume135
192.168.72.136 flume136
192.168.72.136 ganglia

  安装 Ganglia 依赖

sudo apt-get install apache2 mariadb-server php7.0 libapache2-mod-php7.0 php7.0-mbstring php7.0-curl php7.0-zip php7.0-gd php7.0-mysql php7.0-curl php7.0-mcrypt

  安装 Ganglia 的包，分别是监控、服务端和web console

sudo apt-get install ganglia-monitor rrdtool gmetad ganglia-webfrontend

  两次出现弹窗，都选择 no 即可

  拷贝配置文件

sudo cp /etc/ganglia-webfrontend/apache.conf /etc/apache2/sites-enabled/ganglia.conf

  修改 gmetad 配置（/etc/ganglia/gmetad.conf ）

#data_source "my cluster" localhost
data_source "my cluster" 30 192.168.72.136:8649

  修改 gmond 配置（/etc/ganglia/gmond.conf）

...
udp_send_channel {
  #mcast_join = 239.2.11.71
  host = 192.168.72.136
  port = 8649
  ttl = 1
}

udp_recv_channel {
  #mcast_join = 239.2.11.71
  port = 8649
  #bind = 239.2.11.71
}
...

  重启服务

sudo systemctl start ganglia-monitor
sudo systemctl start gmetad
sudo systemctl restart apache2

  查看 Ganglia web console：http://192.168.72.136/ganglia/

  要对 flume agent 加监控，只需要添加启动参数

-Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

  启动 srv02，刷新监控，发现并没有看到 flume132，可以删掉一些 ganglia 文件

cd /var/lib/ganglia/rrds/
sudo rm -rf *

  再重启服务

sudo systemctl restart ganglia-monitor
sudo systemctl restart gmetad
sudo systemctl restart apache2

  可以看到有 flume132 的监控了

  再启动 srv01，并设置 Ganglia 监控，向 srv01（即 flume131）发送一些数据，可以看到 flume 监控指标有数值的变化

  现在将架构图中的所有机器都用 Ganglia 进行监控，按照以下顺序启动

① 启动 master01 的 hdfs 服务

start-dfs.sh

  启动 master01 的 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/master01.conf --conf /app/flume/conf -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

② 启动 srv05 的 HBase 服务

start-hbase.sh

  启动 srv05 的 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/srv05.conf --conf /app/flume/conf -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

③ 启动 srv03 的 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/hdfs-flume.conf --conf /app/flume/conf -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

④ 启动 srv04 的 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/hbase-flume.conf --conf /app/flume/conf -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

⑤ 关闭 flume 服务器，修改 srv02 的配置如下

# 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 = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
a1.sources.r1.ssl = true
a1.sources.r1.keystore = /app/flume/conf/keystore.jks
a1.sources.r1.keystore-password = password
a1.sources.r1.keystore-type = jks

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = flume133
a1.sinks.k1.port = 44444

a1.sinks.k2.type = avro
a1.sinks.k2.hostname = flume134
a1.sinks.k2.port = 44444

# Use a channel which buffers events in memory
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 agent

flume-ng agent --name a1 --conf-file /app/flume/conf/srv02.conf --conf /app/flume/conf -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

⑥ 重启 srv01 flume agent

flume-ng agent --name a1 --conf-file /app/flume/conf/srv01.conf --conf /app/flume/conf -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.72.136:8649 -Dflume.monitoring.pollFrequency=5

  以上步骤全部完成后，可以在 ganglia 监控看到所有的机器
  实现了最初的设计，开心鸭~~