1. 简介

Flume是Cloudera提供的一个分布式、可信任且弹性的系统，用于高效收集、汇聚和移动大规模日志信息从多种不同的数据源到一个集中的数据存储中心。所以可以把Flume当作一个高可用的、高可靠的、分布式的海量日志采集、聚合和传输的系统，支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。 Flume提供了从console、 RPC（Thrift，Avro）、text、tail（UNIX tail）、syslog（支持TCP和UDP等2种模式），exec（命令执行）等数据源上收集数据的能力。

目前Flume是Apache软件基金的顶级项目，有两个可用的分类版本：0.9x和1.x，项目中是使用Apache Flume 1.4版本，即Flume ng，本文讲述的是Flume ng的实现原理及应用。

2. 架构

2.1 数据流模式

一次flume事件定义为一个单元的数据流，该数据流包含一个字节的负荷以及一组可选的字符串属性。一个flume代理是一个JVM进程，用于控制一个组件将外部事件流引导到下一个目的地（hop），如图2.1所示，这是一个Flume Agent。

图2.1采集结构图(1)

一个flume源（source）负责一个外部源，如一个web服务器传递给它的事件。该外部源将它的事件以Flume可以识别的格式发送到flume。例如，一个Avro的flume源可以从Avro客户端接收Avro事件，或者从其他flume代理出口处（sink）接收来自于其他Avro的事件。当一个flume源接收到一个事件时，其将通过一个或者多个通道存储该事件。通道（channel）采用被动存储的形式，即通道会缓存该事件直到该事件被sink组件处理。存储在本地文件系统的文件通道也是一个常用的例子。Sink会将事件从通道中移除并将事件放置到外部数据仓库，例如通过flume HDFS sink将数据放置到HDFS中，或者放置到下一个flume的source中，等待下一个flume进行处理。对于缓存在通道中的事件，source和sink采用异步处理的方式。

2.2 复杂的流动

flume允许用户多次跳转的流，即事件可以在到达目的地之前经过多个flume 代理进行跳转，如下面3个Flume常用的采集结构：

图2.2 采集结构图(2)

图2.3 采集结构图(3)

图2.4 采集结构图（4）

2.3 可靠性

事件被缓存在各个代理的通道中。然后事件会被传送到下一个代理或者最终的存储仓库如HDFS。传送完成后，该事件才会从通道中移除。这是flume保证单次跳转可靠性的方式。 Flume使用事务性的方法来保证事件交互的可靠性。source和sink封装在一个事务的存储和检索中，即事件的放置或者提供由一个事务通过通道来分别提供。这保证了事件集在流中可靠地进行端到端的传递。在多次跳转的情况下，前一跳的sink和下一跳的source都有各自的事务运行来保证数据被安全可靠地存储到下一跳的通道中。

如上图所示，Sink从Channel中获取数据，然后把数据传到另一个Agent中，具体需要经过如下几步：

1)Sink开启事务

2)Sink从Channel中获取数据

3)Sink把数据传给另一个Flume Agent的Source中

4)Source开启事务

5)Source把数据传给Channel

6)Source关闭事务

7)Sink关闭事务

整个处理过程中，如果因为网络中断或者其他原因，在某一步被迫结束了，这个数据会在下一次重新传输。Flume可靠性还体现在数据可暂存上面，如下图所示，当目标不可访问后，数据会暂存在Channel中，等目标可访问之后，再进行传输。

2.4 可还原性

Flume目前支持两种Channel，分别是MemoryChannel和FileChannel，内存通道只是将事件存储在内存中的队列，这是比文件系统速度更快的方式，但是当代理进程失效时，在内存通道中的事件均无法恢复。而FileChannel则支持故障恢复，当FileChannel中有数据没有被传输出去时，Flume就崩溃了，那当Flume重启后，数据会从FileChannel中读取，并进行传输。

3. 核心概念

Agent是flume流的基础部分，一个Agent包含Sources, Channels, Sinks和其他组件，它利用这些组件将events从一个节点传输到另一个节点或最终目的，flume为这些组件提供了配置、生命周期管理、监控支持。这一节将介绍下Flume涉及到的一些比较重要的组件概念，可以分为如下几个部分：

1) Event

2) Sources

3) Channels

4) Sinks

5) 其他组件：Interceptors、Channel Selectors、Sink Processor

从上图中，就可以看出各个组件是如何交互的，数据被封装成Event进行传输，Source用于接收数据，Channel用于暂存数据，Sink用于输出数据，其他组件则是添加一些预处理、分类功能。

3.1 Event

Event是Flume数据传输的基本单元。flume以事件的形式将数据从源头传送到最终的目的。Event由可选的hearders和载有数据的一个byte array构成。

• 载有的数据对flume是不透明的
• Headers是容纳了key-value字符串对的无序集合，key在集合内是唯一的。
• Headers可以在上下文路由中使用扩展

3.2 Source

Source负责接收events或通过特殊机制产生events，并将events批量的放到一个或多个Channels。有event驱动和轮询2种类型的Source

• 不同类型的Source:
  • 和众所周知的系统集成的Sources: Syslog, Netcat
  • 自动生成事件的Sources: Exec, SEQ
  • 用于Agent和Agent之间通信的IPC Sources: Avro、Thrift
• Source必须至少和一个channel关联

通过实现Flume提供的接口，我们可以自定义Source，可以对数据进行一些预处理，具体实现可以参考Flume内置的ThriftSouce类。

3.3 Channel

Channel位于Source和Sink之间，用于缓存进来的events，当Sink成功的将events发送到下一跳的channel或最终目的，events从Channel移除。

• 不同的Channels提供的持久化水平也是不一样的:
  • Memory Channel: volatile
  • File Channel: 基于WAL（预写式日志Write-Ahead Logging）实现
  • JDBC Channel: 基于嵌入Database实现
• Channels支持事务
• 提供较弱的顺序保证
• 可以和任何数量的Source和Sink工作

3.4 Sink

Sink负责将events传输到下一跳或最终目的，成功完成后将events从channel移除。

• 不同类型的Sinks:
  • 存储events到最终目的的终端Sink. 比如: HDFS, HBase
  • 自动消耗的Sinks. 比如: Null Sink
  • 用于Agent间通信的IPC sink: Avro，Thrift
• 必须作用与一个确切的channel

3.5 Interceptor

用于Source的一组Interceptor，按照预设的顺序在必要地方装饰和过滤events。

• 内建的Interceptors允许增加event的headers比如：时间戳、主机名、静态标记等等
• 定制的interceptors可以通过内省event payload（读取原始日志），在必要的地方创建一个特定的headers。

3.6 Channel Selector

Channel Selector允许Source基于预设的标准，从所有Channel中，选择一个或多个Channel

• 内建的Channel Selectors:
  • 复制Replicating: event被复制到相关的channel
  • 复用Multiplexing: 基于hearder，event被路由到特定的channel

3.7 Sink Processor

多个Sink可以构成一个Sink Group。一个Sink Processor负责从一个指定的Sink Group中激活一个Sink。Sink Processor可以通过组中所有Sink实现负载均衡；也可以在一个Sink失败时转移到另一个。

• Flume通过Sink Processor实现负载均衡（Load Balancing）和故障转移（failover）
• 内建的Sink Processors:
  • Load Balancing Sink Processor – 使用RANDOM, ROUND_ROBIN或定制的选择算法
  • Failover Sink Processor
  • Default Sink Processor（单Sink）
• 所有的Sink都是采取轮询（polling）的方式从Channel上获取events。这个动作是通过Sink Runner激活的

Sink Processor充当Sink的一个代理

4. 应用示例

Flume内置了很多实现，所以如果只是想学习下Flume，就只需要通过配置就行了，具体配置如下，并命名问example.conf，放入Flume安装目录FLUME_HOME/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
a1.sources.r1.channels = c1

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Describe the sink
a1.sinks.k1.type = logger
a1.sinks.k1.channel = c1

$bin/flume-ng agent --conf-file example.conf --name a1

然后通过telnet连接，发送数据”hello world!”，如下：

$ telnet localhost 44444

Trying 127.0.0.1...

Connected to localhost.localdomain (127.0.0.1).

Escape character is '^]'.

Hello world! <ENTER>

OK

5. 小结

Flume作为一个海量日志采集、聚合、传输系统，本身架构比较简单，但就是因为简单，所以才稳定、灵活。本文只是介绍了下Flume的实现原理和怎样使用Flume，如果有兴趣可以去看下Flume源码，Flume比较好的地方不仅在于它给你提供了很多现成的工具，还在于它的接口封装的比较好，我们可以根据它提供的接口，根据自身业务，实现自定义的Source、Channel、Sink或者其他组件的编写。

6. 参考资料

1. 官网User Guide http://flume.apache.org/FlumeUserGuide.html
2. 官网Developer Guidehttp://flume.apache.org/FlumeDeveloperGuide.html
3. Flume ng Performance Measuermentshttps://cwiki.apache.org/confluence/display/FLUME/Flume+NG+Performance+Measurements