前提：
1)五台虚拟机（三台也可以）
2)flume单节点测试并学会
3)hadoop集群搭建完成
Flume NG集群，架构图
Flume的存储可以支持多种，这里只列举了HDFS
角色分配
| 名称 | HOST | 角色 |
|---|---|---|
| Agent1 | chun1 | Web Server |
| Agent2 | chun2 | Web Server |
| Agent3 | chun3 | Web Server |
| Collector1 | chun4 | AgentMstr1 |
| Collector2 | chun5 | AgentMstr1 |
表中所示，Agent1，Agent2，Agent3数据分别流入到Collector1和Collector2，Flume NG本身提供了Failover机制，可以自动切换和恢复。在上图中，有3个产生日志服务器分布在不同的机房，要把所有的日志都收集到一个集群中存储。下 面我们开发配置Flume NG集群
配置
在单点Flume中（这里介绍了单点的配置），基本配置都完成了，我们只需要新添加两个配置文件，它们是agent.properties和collector.properties，其配置内容如下所示：
agent配置
(根据自己需求把source读的路径（r1.command ）和要配置的collector的主机名修改也就是k1和k2的hostname)
[root@chun1 flume-1.9.0-bin]# vi conf/agent.properties#agent1 name
agent1.channels = c1
agent1.sources = r1
agent1.sinks = k1 k2#set gruop
agent1.sinkgroups = g1#set channel
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 1000
agent1.channels.c1.transactionCapacity = 100agent1.sources.r1.channels = c1
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /usr/local/flume-1.9.0/job/log/test.log
agent1.sources.r1.interceptors = i1 i2
agent1.sources.r1.interceptors.i1.type = static
agent1.sources.r1.interceptors.i1.key = Type
agent1.sources.r1.interceptors.i1.value = LOGIN
agent1.sources.r1.interceptors.i2.type = timestamp# set sink1
agent1.sinks.k1.channel = c1
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = chun4
agent1.sinks.k1.port = 52020# set sink2
agent1.sinks.k2.channel = c1
agent1.sinks.k2.type = avro
agent1.sinks.k2.hostname = chun5
agent1.sinks.k2.port = 52020#set sink group
agent1.sinkgroups.g1.sinks = k1 k2#set failover
agent1.sinkgroups.g1.processor.type = failover
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
agent1.sinkgroups.g1.processor.maxpenalty = 10000
修改后把flume发送给chun1，chun2，chun3，chun4，chun5( 发送后chun1，chun2，chun3不需要修改)
（chun4，chun5把刚才创建的agent.properties删除，添加一个collector.properties 并加入以下内容）
collector配置
记得把主机名改掉
[root@chun4 conf]# vi collector.properties #set Agent name
a1.sources = r1
a1.channels = c1
a1.sinks = k1#set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = chun4 //chun5的此处要改
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = chun4 //chun5的此处要改
a1.sources.r1.channels = c1#set sink to hdfs
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/home/hdfs/flume/logdfs
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d
运行 (先启动两个collector然后在启动三个agent)
在4，5上运行
cd /usr/local/flume-1.9.0bin/flume-ng agent -n a1 -c conf -f conf/collector.properties -Dflume.root.logger=DEBUG,console
在1，2，3上运行
cd /usr/local/flume-1.9.0bin/flume-ng agent -n agent1 -c conf -f conf/agent.properties -Dflume.root.logger=DEBUG,console
插入数据
往test.txt里插入数据
代码意思：没0.5秒循环插入chun-chun-chun
while true
> do
> echo 'chun-chun-chun' >> /usr/local/flume-1.9.0/job/log/test.log
> sleep 0.5
> done
查看 （hdfs的web端查看）
这时你会发现只有flume-ng1下有数据：说明是先往chun4上传
然后把chun4的进程杀死，就会发现数据开始往chun5传
然后再次打开（再次启动报错请看）数据又到chun4了（数据会先往权重高的传输）
（配置文件里有设置权重
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
）可以根据自己需求设置
