Flume官方文档的理解和配置操作
摘要
Flume是一个分布式的，可靠的，并且具有高可用性的一个服务系统，用于高效地收集、聚合、移动一个大数量的日志文件， 它有一个基于数据流类型的简单且灵活的的架构。 由于它的容错可调 机制和许多故障转移和恢复机制，使得这个架构具有较高的容错性和健壮性。 它使用一个简单的可扩展的数据模型,可以让它在线分析应用程序。
一、概述
Apache Flume是一个分布式的、可靠的、高可用性的一个系统，用于从许多不同来源的一个集中的数据存储中高效的收集、聚合、移动一个大数据量的日志文件（分布式的海量日志的高效收集、聚合、移动/传输的一个框架），特点：简单灵活，是基于一个流式数据处理框架。
二、系统需求
1.Java运行环境（jre）1.8版本及以上
三、 体系结构
数据流模型
理解：
Source：数据从哪里去。去源头读数据，写到Channel中
四、复杂的流
Flume允许用户构建多重流动的数据流， 在这些流被传输到最后重点是，用户可以它进行多重的控制。 它还允许扇入 和扇出流，允许进行路线的选择以便其中一条路线出现故障。
五、可靠性
一个数据的代理是在一个通道中， 然后数据被传输到下一个代理机或者终端存储器 例如HDFS。数据仅在他们存到下一个代理机的通道或者到达终端存储器时会被当前的通道移除。这便是为何flume提供的简答的端到端交付数据，信息传输拥有可靠性。
六、可恢复性
数据存储在通道中，出现故障时可以随时回复，Flume提供了一个持久耐用的文件通道有本地文件系统支持，还有一个记忆通道，简单地将数据存储在记忆队列中，可以加快数据的传输但是任何一个数据遗留在通道中时代理机出现故障，数据无法恢复。
七、配置
7.1 配置一个flume
Flume代理配置存在一个本地配置文件中，这是一个遵循Java属性文件格式的文本文件，可以指定一个或者多个代理的配置位相同的配置文件，配置文件包含每个数据源的性质，说明了他们是如何连接在一起形成数据流的。
7.2配置单个组件
每个组件（source，channel，sink）有一个名称，类型，并且需要设定特定属性的类型和实例化。例如，一个Avro源需要一个主机名（或IP地址）和接受数据的端口号。一个内存通道拥有存一个最大队列的能力，一个HDFS sink需要知道系统URI，按照路径创建文件，设置文件的属性等。所有这些组件的属性都需要有Flume代理机设定。
7.3把组件联合在一起
代理机需要知道每个组件加载是如何连接并形成流。这是通过列出每一个source，sink和channel在代理机中的命名，然后具体地连接这三个组件。例如，一个来自AVro source的代理数据流命名为avroWeb 发送到 名为hdfs-clust1的HDFS sink，这次传输通过名字为file-channel的通道。配置文件将包含这些的名字组件和file-channel avroWeb作为一个公共信道提供给avroWeb源和hdfs-cluster1 sink。
7.4设置一个代理
一个代理使用shell脚本位于flume的bin目录下的名为flume-ng的文件，你需要制定具体的代理名称，配置目录，在命令行配置文件：
现在，代理开始运行source和sink中的配置属性文件
7.5一个简单的例子
这里，我们给出一个示例配置稳健，描述一个单节点flume部署，该配置允许用户生成数据和随后的日志控制台。
# example.conf: A single-node Flume configuration
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
这个配置定义了一个单一的代理叫a1.a1源在端口44444上获取数据，一个通道在内存中缓冲数据，sink发送日志数据到控制台。配置文件名称和各个组件，然后描述了他们的类型和配置参数。给定配置文件定义的叫代理机，当一个给定的flume过程启动标志这名为代理机表明给定的配置文件可用，我们就可以允许flume了。
有了这个配置文件，我们可以启动Flume如下:
$ bin/flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=INFO,console
需要注意的是，在一个完成的配置中通常包括一个选项：–conf=. The 目录下的包括一个shell脚本flume-enc.Sh和一个潜在的log4j文件。例如，我们通过一个java option强制访问日志控制台 但是我们没有自定义环境脚本。
$ telnet localhost 44444
Trying 127.0.0.1...
Connected to localhost.localdomain (127.0.0.1).
Escape character is '^]'.
Hello world! <ENTER>
OK
最终flume中孤单将在一个日志信息输出数据
至此我们就成功配置了一个flume agent。
7.6自己配置的示例
Windows下配置flume
步骤1：安装java ，配置环境变量。
# 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
步骤4：打开cmd进入到apache-flume-1.7.0-bin\bin目录下，运行如下命令
八、有环境变量配置文件
Flume有能力配置环境变量，举个例子：
注意：
它仅仅对value值进行操作，对key值无效。（仅是在右边的等同于对配置行做标记）
九、记录原始数据
记录原始数据流是通过接收数据传输路径中的数据，这不是很多开发环境中预期的操作，因为这可能会导致敏感数据泄露或者影响安全相关配置，例如密码和flume日志文件。在编码时，Flume不会记录这些信息。从另一个方面来看，如果数据传输路径故障，Flume会尝试为调试解决故障提供线索。
十、Zookeeper的基础配置
Flume支持通过Zookeeper代理配置，这是一个新开发的特色。配置文件需要由Zookeeper来上传，在一个可配置的前缀下。配置文件存储在Zookeeper节点数据中。下面就是Zookeeper节点数怎么样来代理a1和a2配置文件。
十一.安装第三方插件
Flume有一个完整的plugin-based 架构。当Flume用sources，channels，sinks，serializers之类的外部资源传输数据时，大量存在实现flume分离式传输。
十二、多智能体流设置
十三、多重复杂流
Flume支持多路传输数据流至一个或者多个目标地。这是通过定义流多路复用器实现的，该多路复用器可以复制或有选择地将事件数据传输到一个或多个通道。
十四、代理配置
正如前面提到的部分,Flume代理配置是读取一个文件,就像java属性文件格式层次属性设置。
十五丶整合
日志收集中的一个非常常见的场景是大量的日志生成客户机将数据发送给连接到存储子系统的几个客户代理。例如，从几百个web服务器上收集的日志发送到12个代理，这些代理将写入HDFS集群。
十六、定义流
要在单个代理中定义流，需要通过通道链接源接收。您需要列出给定代理的源、接收器和通道，然后将源和接收器指向一个通道。源实例可以指定多个通道，但是接收器实例只能指定一个通道。格式如下:
十七、 配置单个组件
定义流之后，需要设置每个源、接收器和通道的属性。这是按照相同的层次命名空间方式进行的，您在其中为每个组件特定的属性设置组件类型和其他值:
十八 、在代理中添加多个流
一个单独的Flume代理可以包含几个独立的流。您可以在配置中列出多个源、接收器和通道。这些组件可以连接成多个流:
十九、配置多重代理流
设置一个多层的流,你就需要一个avro /thrift第一跳指着接收器sink avro /thrift的下一个hop。这将导致第一个Flume代理将事件转发给下一个Flume代理。例如，如果您正在使用avro客户端定期向本地Flume代理发送文件(每个事件一个文件)，那么这个本地代理可以将文件转发给另一个挂载用于存储的代理。
二十、扇出流
正如前面讨论的，Flume支持将流从一个源分散到多个通道。扇出有两种模式，复制和多路复用。在复制流中，事件被发送到所有配置的通道。在多路复用的情况下，事件只被发送到限定通道的一个子集。要扇出流，需要指定源的通道列表和扇出源的策略。这是通过添加可以复制或多路复用的通道“选择器”来实现的。然后进一步指定选择规则，如果它是一个多路复用器。如果没有指定选择器，那么默认情况下它是复制的:
二十一 、Flume组件
下图展示了Flume各组件功能及扩展的展示
21.1 Source
NetCat Source：绑定的端口（tcp、udp），将流经端口的每一个文本行数据作为Event输入；
type：source的类型，必须是netcat。
bind：要监听的(本机的)主机名或者ip。此监听不是过滤发送方。一台电脑不是说只有一个IP。有多网卡的电脑，对应多个IP。
port：绑定的本地的端口。
Avro Source：监听一个avro服务端口，采集Avro数据序列化后的数据；
type：avrosource的类型，必须是avro。
bind：要监听的(本机的)主机名或者ip。此监听不是过滤发送方。一台电脑不是说只有一个IP。有多网卡的电脑，对应多个IP。
port：绑定的本地的端口。
Exec Source：于Unix的command在标准输出上采集数据；
type:source的类型：必须是exec。
command：要执行命令。
现在选择其中的Avro Source进行说明：
监听Avro端口并从外部Avro客户端流接收事件。当与另一个(前一个hop）Flume代理上的内置Avro接收器配对时，它可以创建分层集合拓扑。所需的属性是blod 。
a1.sources = r1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4141
ipFilterRules的例子：
ipFilterRules defines N netty ipFilters separated by a comma a pattern rule must be in this format.
<’allow’ or deny>:<’ip’ or ‘name’ for computer name>:<pattern> or allow/deny:ip/name:pattern
example: ipFilterRules=allow:ip:127.*,allow:name:localhost,deny:ip:*
请注意，匹配的第一个规则将应用于本地主机上的客户端所显示的示例。
21.2 sink
HDFS Sink：将数据传输到hdfs集群中。
type：sink的类型 必须是hdfs。
hdfs.path：hdfs的上传路径。
hdfs.filePrefix：hdfs文件的前缀。默认是:FlumeData
hdfs.rollInterval:间隔多久产生新文件，默认是:30（秒） 0表示不以时间间隔为准。
hdfs.rollSize：文件到达多大再产生一个新文件，默认是:1024（bytes）0表示不以文件大小为准。
hdfs.rollCount：event达到多大再产生一个新文件，默认是:10（个）0表示不以event数目为准。
hdfs.batchSize：每次往hdfs里提交多少个event，默认为100
hdfs.fileType：hdfs文件的格式主要包括：SequenceFile, DataStream ,CompressedStream，如果使用了CompressedStream就要设置压缩方式。
hdfs.codeC：压缩方式：gzip, bzip2, lzo, lzop, snappy
注：%{host}可以使用header的key。以及%Y%m%d来表示时间，但关于时间的表示需要在header里有timestamp这个key。
Logger Sink将数据作为日志处理（根据flume中的设置的日志方式来显示）
要在控制台显示在运行agent的时候加入：-Dflume.root.logger=INFO,console 。
type：sink的类型：必须是 logger。
maxBytesToLog：打印body的最长的字节数 默认为16
Avro Sink：数据被转换成Avro Event，然后发送到指定的服务端口上。
type：sink的类型：必须是 avro。
hostname：指定发送数据的主机名或者ip
port：指定发送数据的端口
下面以HDFS sink为例进行说明
这个接收器将事件写入Hadoop分布式文件系统(HDFS)。它目前支持创建文本和序列文件。它支持两种文件类型的压缩。可以根据数据的运行时间、大小或事件的数量周期性地滚动文件(关闭当前文件并创建新文件)。它还根据事件发生的时间戳或机器等属性对数据进行分段或分区。HDFS目录路径可能包含格式化转义序列，这些转义序列将被HDFS接收器替换为生成一个目录/文件名来存储事件。使用这个接收器需要安装hadoop，以便Flume可以使用hadoop jar与HDFS集群通信。注意，需要一个支持sync()调用的Hadoop版本。 bold 。
a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
上述配置将把时间戳缩短到最后10分钟。例如，2012年6月12日上午11:54:34的事件将导致hdfs路径变成/flume/events/2012-06-12/1150/00。
21.3. channel
Memory Channel使用内存作为数据的存储。
Type channel的类型：必须为memory
capacity：channel中的最大event数目
transactionCapacity：channel中允许事务的最大event数目
File Channel 使用文件作为数据的存储
Type channel的类型：必须为 file
checkpointDir ：检查点的数据存储目录
dataDirs ：数据的存储目录
transactionCapacity：channel中允许事务的最大event数目
Spillable Memory Channel 使用内存作为channel超过了阀值就存在文件中
Type channel的类型：必须为SPILLABLEMEMORY
memoryCapacity：内存的容量event数
overflowCapacity：数据存到文件的event阀值数
checkpointDir：检查点的数据存储目录
dataDirs：数据的存储目录
下面以Memory Channel为例进行说明：
事件存储在内存队列中，具有可配置的最大大小。对于需要更高吞吐量并准备在代理失败时丢失阶段数据的流来说，这是理想的。所需的属性是bold 。
a1.channels = c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000
21.4. Flume Channel Selector
Multiplexing Channel Selector 根据header的key的值分配channel
selector.type 默认为replicating
selector.header：选择作为判断的key
selector.default：默认的channel配置
selector.mapping.*：匹配到的channel的配置
下面以Multiplexing Channel Selector为例进行说明
所需属性是bold 。
a1.sources = r1
a1.channels = c1 c2 c3
a1.sources.r1.selector.type = replicating
a1.sources.r1.channels = c1 c2 c3
a1.sources.r1.selector.optional = c3
在上面的配置中，c3是一个可选的通道。对c3的写入失败被简单地忽略。由于c1和c2没有被标记为可选的，因此不向这些通道写入将导致事务失败。
21.5 Flume Sink Processor
Default sink processor
21.6 Event Serializer
BodyText Serializer
21.7 Interceptor
Timestamp Interceptor 时间戳拦截器 在header里加入key为timestamp，value为当前时间。
type：拦截器的类型，必须为timestamp
preserveExisting：如果此拦截器增加的key已经存在，如果这个值设置为true则保持原来的值，否则覆盖原来的值。默认为false
Host Interceptor 主机名或者ip拦截器，在header里加入ip或者主机名
type：拦截器的类型，必须为host
preserveExisting：如果此拦截器增加的key已经存在，如果这个值设置为true则保持原来的值，否则覆盖原来的值。默认为false
useIP：如果设置为true则使用ip地址，否则使用主机名，默认为true
hostHeader：使用的header的key名字，默认为host
Static Interceptor 静态拦截器，是在header里加入固定的key和value。
type：avrosource的类型，必须是static。
preserveExisting:如果此拦截器增加的key已经存在，如果这个值设置为true则保持原来的值，否则覆盖原来的值。默认为false
key:静态拦截器添加的key的名字
value:静态拦截器添加的key对应的value值
下面我们以Timestamp Interceptor为例进行展示
这个拦截器将插入到事件头中，即它处理事件的时间。这个拦截器插入一个标头，标头的键时间戳(或标头属性指定的时间戳)的值是相关的时间戳。如果已经存在于配置中，此拦截器可以保存现有的时间戳。
a1.sources = r1
a1.channels = c1
a1.sources.r1.channels = c1
a1.sources.r1.type = seq
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp
一些其他的例子
Example 1：
a1.sources.r1.interceptors.i1.regex = (\\d):(\\d):(\\d)
a1.sources.r1.interceptors.i1.serializers = s1 s2 s3
a1.sources.r1.interceptors.i1.serializers.s1.name = one
a1.sources.r1.interceptors.i1.serializers.s2.name = two
a1.sources.r1.interceptors.i1.serializers.s3.name = three
提取的事件将包含相同的主体，但以下标题将被添加1 =>1,2 =>2,3 =>3
Example2：
a1.sources.r1.interceptors.i1.regex = ^(:\\n)(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)
a1.sources.r1.interceptors.i1.serializers = s1
a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer
a1.sources.r1.interceptors.i1.serializers.s1.name = timestamp
a1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm
提取的事件将包含相同的主体，但将添加以下头部:timestamp=>1350611220000
Component Summary
下面表格包含了所有组件名称信息
文件通道完整性工具在文件通道中验证单个事件的完整性，并清除损坏的事件。
$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir
其中datadir是要验证的数据目录的逗号分隔列表。
事件验证器工具可用于以应用程序特定的方式验证文件通道事件。该工具对每个事件应用用户提供程序验证登录，并删除不确认逻辑的事件。
$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir -e org.apache.flume.MyEventValidator -DmaxSize 2000
其中datadir是要验证的数据目录的逗号分隔列表。
public static class MyEventValidator implements EventValidator {
private int value = 0;
private MyEventValidator(int val) {
value = val;
}
@Override
public boolean validateEvent(Event event) {
return event.getBody() <= value;
}
public static class Builder implements EventValidator.Builder {
private int sizeva lidator = 0;
@Override
public EventValidator build() {
return new DummyEventVerifier(sizeva lidator);
}
@Override
public void configure(Context context) {
binaryValidator = context.getInteger("maxSize");
}
}
