这种模式，有两种方式，一种是用来复制（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。

实现load balance功能

Load balancing Sink Processor能够实现load balance功能，上图Agent1是一个路由节点，负责将Channel暂存的Event均衡到对应的多个Sink组件上，而每个Sink组件分别连接到一个独立的Agent上，示例配置，如下所示：

a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 k3 a1.sinkgroups.g1.processor.type = load_balance a1.sinkgroups.g1.processor.backoff = true a1.sinkgroups.g1.processor.selector = round_robin a1.sinkgroups.g1.processor.selector.maxTimeOut=10000 实现failover能

Failover Sink Processor能够实现failover功能，具体流程类似load balance，但是内部处理机制与load balance完全不同：Failover Sink Processor维护一个优先级Sink组件列表，只要有一个Sink组件可用，Event就被传递到下一个组件。如果一个Sink能够成功处理Event，则会加入到一个Pool中，否则会被移出Pool并计算失败次数，设置一个惩罚因子，示例配置如下所示：

a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 k3 a1.sinkgroups.g1.processor.type = failover a1.sinkgroups.g1.processor.priority.k1 = 5 a1.sinkgroups.g1.processor.priority.k2 = 7 a1.sinkgroups.g1.processor.priority.k3 = 6 a1.sinkgroups.g1.processor.maxpenalty = 20000

4. 安装和使用

Flume 的 rpm 安装方式很简单，这里不做说明。

示例1： avro 数据源

安装成功之后，在 /etc/flume/conf 目录创建f1.conf 文件，内容如下:

agent-1.channels.ch-1.type = memory agent-1.sources.avro-source1.channels = ch-1 agent-1.sources.avro-source1.type = avro agent-1.sources.avro-source1.bind = 0.0.0.0 agent-1.sources.avro-source1.port = 41414 agent-1.sources.avro-source1.threads = 5 agent-1.sinks.log-sink1.channel = ch-1 agent-1.sinks.log-sink1.type = logger agent-1.channels = ch-1 agent-1.sources = avro-source1 agent-1.sinks = log-sink1

关于 avro-source 配置说明，请参考 avro-source

接下来启动 agent：

$ flume-ng agent -c /etc/flume-ng/conf -f /etc/flume-ng/conf/f1.conf -Dflume.root.logger=DEBUG,console -n agent-1

参数说明：

-n 指定agent名称-c 指定配置文件目录-f 指定配置文件-Dflume.root.logger=DEBUG,console 设置日志等级

下面可以启动一个 avro-client 客户端生产数据：

$ flume-ng avro-client -c /etc/flume-ng/conf -H localhost -p 41414 -F /etc/passwd -Dflume.root.logger=DEBUG,console

示例2：spooldir 数据源

在 /etc/flume/conf 目录创建 f2.conf 文件，内容如下:

agent-1.channels = ch-1 agent-1.sources = src-1 agent-1.channels.ch-1.type = memory agent-1.sources.src-1.type = spooldir agent-1.sources.src-1.channels = ch-1 agent-1.sources.src-1.spoolDir = /root/log agent-1.sources.src-1.fileHeader = true agent-1.sinks.log-sink1.channel = ch-1 agent-1.sinks.log-sink1.type = logger agent-1.sinks = log-sink1

关于 Spooling Directory Source 配置说明，请参考 Spooling Directory Source

接下来启动 agent：

$ flume-ng agent -c /etc/flume-ng/conf -f /etc/flume-ng/conf/f2.conf -Dflume.root.logger=DEBUG,console -n agent-1

然后，手动拷贝一个文件到 /root/log 目录，观察日志输出以及/root/log 目录下的变化。

示例3：spooldir 数据源，写入 hdfs

在 /etc/flume/conf 目录创建 f3.conf 文件，内容如下:

agent-1.channels.ch-1.type = file agent-1.channels.ch-1.checkpointDir= /root/checkpoint agent-1.channels.ch-1.dataDirs= /root/data agent-1.sources.src-1.type = spooldir agent-1.sources.src-1.channels = ch-1 agent-1.sources.src-1.spoolDir = /root/log agent-1.sources.src-1.deletePolicy= never agent-1.sources.src-1.fileHeader = true agent-1.sources.src-1.interceptors =i1 agent-1.sources.src-1.interceptors.i1.type = timestamp agent-1.sinks.sink_hdfs.channel = ch-1 agent-1.sinks.sink_hdfs.type = hdfs agent-1.sinks.sink_hdfs.hdfs.path = hdfs://cdh1:8020/user/root/events/%Y-%m-%d agent-1.sinks.sink_hdfs.hdfs.filePrefix = logs agent-1.sinks.sink_hdfs.hdfs.inUsePrefix = . agent-1.sinks.sink_hdfs.hdfs.rollInterval = 30 agent-1.sinks.sink_hdfs.hdfs.rollSize = 0 agent-1.sinks.sink_hdfs.hdfs.rollCount = 0 agent-1.sinks.sink_hdfs.hdfs.batchSize = 1000 agent-1.sinks.sink_hdfs.hdfs.writeFormat = text agent-1.sinks.sink_hdfs.hdfs.fileType = DataStream #agent-1.sinks.sink_hdfs.hdfs.fileType = CompressedStream#agent-1.sinks.sink_hdfs.hdfs.codeC = lzop agent-1.channels = ch-1 agent-1.sources = src-1 agent-1.sinks = sink_hdfs

关于 HDFS Sink配置说明，请参考 HDFS Sink

说明：

通过 interceptors 往 header 里添加 timestamp，这样做，可以在 hdfs.path 引用系统内部的时间变量或者主机的 hostname。通过设置 hdfs.inUsePrefix，例如设置为 .时，hdfs 会把该文件当做隐藏文件，以避免在 mr 过程中读到这些临时文件，引起一些错误如果使用 lzo 压缩，则需要手动创建 lzo 索引，可以通过修改 HdfsSink 的代码，通过代码创建索引FileChannel 的目录最好是和 spooldir 的数据目录处于不同磁盘。

示例4：spooldir 数据源，写入 HBase

关于 HBase Sink 配置说明，请参考 HBase Sink

5. 开发相关

5.1 编译源代码

从 github 下载源代码并编译：

$ git clone git@github.com:cloudera/flume-ng.git -b cdh4-1.4.0_4.7.0 $ cd flume-ng $ mvn install -DskipTests -Phadoop-2

如果提示找不到 hadoop-test 的 jar 包，则修改 pom.xml 中的版本，如改为 2.0.0-mr1-cdh4.7.0，具体版本视你使用的分支版本而定，我这里是 cdh4.7.0。

如果提示找不到 uanodeset-parser 的 jarb，则在 pom.xml 中添加下面仓库：

<repository> <id>tempo-db</id> <url>http://maven.tempo-db.com/artifactory/list/twitter/ </url> <snapshots> <enabled>false</enabled> </snapshots> </repository>

6. 最佳实践

参考基于Flume的美团日志收集系统(一)架构和设计，列出一些最佳实践：

模块命名规则：所有的 Source 以 src 开头，所有的 Channel 以 ch 开头，所有的 Sink 以 sink 开头；模块之间内部通信统一使用 Avro 接口；将日志采集系统系统分为三层：Agent 层，Collector 层和 Store 层，其中 Agent 层每个机器部署一个进程，负责对单机的日志收集工作；Collector 层部署在中心服务器上，负责接收Agent层发送的日志，并且将日志根据路由规则写到相应的 Store 层中；Store 层负责提供永久或者临时的日志存储服务，或者将日志流导向其它服务器。扩展 MemoryChannel 和 FileChannel ，提供 DualChannel 的实现，以提供高吞吐和大缓存监控 collector HdfsSink写数据到 hdfs 的速度、FileChannel 中拥堵的 events 数量，以及写 hdfs 状态（查看是否有 .tmp 文件生成）

美团对 flume 的改进代码见 github：https://github.com/javachen/mt-flume。

监听端口 #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=Master a1.sources.r1.port=44444 #Deacribe 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