关于使用Flume（NG）打造日志系统的分享与思考

可以将多个Agent顺序连接起来，将最初的数据源进过收集，存储到最终的存储系统中。这是最简单的情况，一般情况下，应该控制这种顺序连接下Agent的数量，因为数据流经的路径变长了，如果不考虑failover的话，出现故障将影响到整个Flow上的Agent的收集服务。

这种情况应用的场景比较多，比如要收集web网站的用户行为日志，Web网站为了应用的可用性使用负载均衡的集群模式，每个节点都会产生用户行为日志，可以为每个节点都配置一个Agent来单独收集日志数据，然后多个Agent将数据汇总到一个用来存储数据的存储系统，如：HDFS。

这种模式，有两种方式，一种用来复制（replication），另一种用来分流（Multiplexing）。

• Replication方式，可以将最前端的数据源复制多份，分别传递到多个channel中，每个channel接收到的数据都是相同的，配置格式，如下所示：

# List the sources, sinks and channels for the agent
<Agent>.sources = <Source1>
<Agent>.sinks = <Sink1> <Sink2>
<Agent>.channels = <Channel1> <Channel2>

# set list of channels for source (separated by space)
<Agent>.sources.<Source1>.channels = <Channel1> <Channel2>

# set channel for sinks
<Agent>.sinks.<Sink1>.channel = <Channel1>
<Agent>.sinks.<Sink2>.channel = <Channel2>

<Agent>.sources.<Source1>.selector.type = replicating

上面指定了selector的type的值为replication，其他的配置没有指定，使用的Replication方式，Source1会将数据分别存储到Channel1和Channel2，这两个channel里面存储的数据是相同的，然后数据被传递到Sink1和Sink2。

• Multiplexing方式，selector可以根据header的值来确定数据传递到哪一个channel，配置格式，如下所示：

# Mapping for multiplexing selector
<Agent>.sources.<Source1>.selector.type = multiplexing
<Agent>.sources.<Source1>.selector.header = <someHeader>
<Agent>.sources.<Source1>.selector.mapping.<Value1> = <Channel1>
<Agent>.sources.<Source1>.selector.mapping.<Value2> = <Channel1> <Channel2>
<Agent>.sources.<Source1>.selector.mapping.<Value3> = <Channel2>
#...

<Agent>.sources.<Source1>.selector.default = <Channel2>

上面selector的type的值为multiplexing，同时配置selector的header信息，还配置了多个selector的mapping的值，即header的值：如果header的值为Value1、Value2，数据从Source1路由到Channel1；如果header的值为Value2、Value3，数据从Source1路由到Channel2。

Source类型	说明
Avro Source	支持Avro协议（实际上是Avro RPC）,内置支持
Thrift Source	支持Thrift协议，内置支持
Exec Source	基于Unix的command在标准输出上生产数据
JMS Source	从JMS(消息，主题)中读取数据
Spooling Directory Source	监控指定目录内数据的变更
Twitter 1% firehose Source	通过API持续下载Twitter数据，试验性质
Netcat Source	监控某个端口，将流经端口的每个文本行数据作为Event输入
Sequence Generator Source	序列生成器数据源，生产序列数据
Syslog Source	读取syslog数据，产生Event，支持UDP和TCP两种协议
HTTP Source	基于HTTP POST或GET方式的数据源，支持JSON，BLOB表示格式
Legacy Source	兼容老的Flume OG中Source（0.9.x版本）

Sink类型	说明
HDFS Sink	数据写入HDFS
logger Sink	数据写入日志文件
Avro Sink	数据被转换成Avro Event，然后发送到配置的RPC端口上
Thrift Sink	数据被转换成Thrift Event，然后发送到配置的RPC端口上
IRC Sink	数据在IRC上进行回收
File Roll Sink	存储数据到本地文件系统
Null Sink	丢弃所有数据
HBase Sink	数据写入Hbase数据库
Morphline Solr Sink	数据发送到Solr搜素服务器（集群）
ElasticSearch Sink	数据发送到Elastic Search搜素服务器
Kite Dataset Sink	写数据到Kite Dataset中
Custom Sink	自定义Sink实现

Channel类型	说明
Memory Channel	Event数据存储在内存中
JDBC Channel	Event数据存储在持久化存储中，当前Flume Channel内置支持Derby
File Channel	Event数据存储在磁盘文件中
Spillable Memory Channel	Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件
Pseudo Transaction Channel	测试用途
Custom Channel	自定义Channel实现

1. 安装java，配置环境变量。
2. 安装flume，下载地址，下载后直接解压即可。

• 创建配置文件：在解压后的文件apache-flume-1.7.0-bin/conf下创建一个example.conf，内容如下。

# 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

• 打开cmd进入到apache-flume-1.7.0-bin\bin目录下，运行如下命令。

flume-ng.cmd  agent -conf ../conf  -conf-file ../conf/example.conf  -name a1  -property flume.root.logger=INFO,console

• 另外打开一个cmd窗口，运行如下命令。

telnet localhost 44444