一、什么是 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/）查看。