Flume NG 学习笔记（六）Selector（复用与复制）测试
版权声明：本文为博主原创文章，未经博主允许不得转载。
目录(?)[+]
学习心得（三）流配置中介绍多路复用流的时候，有说到Flume支持从一个源发送事件到多个通道中，这被称为事件流的复用。这里需要在配置中定义事件流的复制/复用，选择1个或者多个通道进行数据流向。
而关于selector配置前面也讲过：
<Agent>.sources.<Source1>.selector.type= replicating
这个源的选择类型为复制。这个参数不指定一个选择的时候，默认情况下它复制
复用则是麻烦一下，流的事情是被筛选的发生到不同的渠道，需要指定源和扇出通道的规则，感觉与case when 类似。
复用的参数为：
<Agent>.sources.<Source1>.selector.type= multiplexing
这里需要配置1个代理作为源发送与2个代理作为接受复制事件，共3个flume配置
首先是作为源发送的代理配置
[html]view plaincopy
#配置文件：replicate_source_case11.conf
#Namethecomponentsonthisagent
a1.sources=r1
a1.sinks=k1k2
a1.channels=c1c2
#Describe/configurethesource
a1.sources.r1.type=syslogtcp
a1.sources.r1.port=50000
a1.sources.r1.host=192.168.233.128
a1.sources.r1.selector.type=replicating
a1.sources.r1.channels=c1c2
#Describethesink
a1.sinks.k1.type=avro
a1.sinks.k1.channel=c1
a1.sinks.k1.hostname=192.168.233.129
a1.sinks.k1.port=50000
a1.sinks.k2.type=avro
a1.sinks.k2.channel=c2
a1.sinks.k2.hostname=192.168.233.130
a1.sinks.k2.port=50000
#Useachannelwhichbufferseventsinmemory
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.channels.c2.type=memory
a1.channels.c2.capacity=1000
a1.channels.c2.transactionCapacity=100
这里设置了2个channels与2个sinks，那么我们也要设置2个sinks对应的代理配置：
下面是第一个接受复制事件代理配置
[html]view plaincopy
#配置文件：replicate_sink1_case11.conf
#Namethecomponentsonthisagent
a2.sources=r1
a2.sinks=k1
a2.channels=c1
#Describe/configurethesource
a2.sources.r1.type=avro
a2.sources.r1.channels=c1
a2.sources.r1.bind=192.168.233.129
a2.sources.r1.port=50000
#Describethesink
a2.sinks.k1.type=logger
a2.sinks.k1.channel=c1
#Useachannelwhichbufferseventsinmemory
a2.channels.c1.type=memory
a2.channels.c1.capacity=1000
a2.channels.c1.transactionCapacity=100
下面是第二个接受复制事件代理配置：
[html]view plaincopy
#配置文件：replicate_sink2_case11.conf
#Namethecomponentsonthisagent
a3.sources=r1
a3.sinks=k1
a3.channels=c1
#Describe/configurethesource
a3.sources.r1.type=avro
a3.sources.r1.channels=c1
a3.sources.r1.bind=192.168.233.130
a3.sources.r1.port=50000
#Describethesink
a3.sinks.k1.type=logger
a3.sinks.k1.channel=c1
#Useachannelwhichbufferseventsinmemory
a3.channels.c1.type=memory
a3.channels.c1.capacity=1000
a3.channels.c1.transactionCapacity=100
#敲命令
首先先启动2个接受复制事件代理，如果先启动源发送的代理，会报他找不到sinks的绑定，因为2个接事件的代理还未起来。
flume-ng agent -cconf -f conf/replicate_sink1_case11.conf -n a1-Dflume.root.logger=INFO,console
flume-ng agent -cconf -f conf/replicate_sink2_case11.conf -n a1-Dflume.root.logger=INFO,console
在启动源发送的代理
flume-ng agent -cconf -f conf/replicate_source_case11.conf -n a1-Dflume.root.logger=INFO,console
启动成功后
打开另一个终端输入，往侦听端口送数据
echo "hello looklook5"| nc 192.168.233.128 50000
#在启动源发送的代理终端查看console输出
可以看到他的正常启动以及发送数据成功
#在启动源第一个接事件的代理终端查看console输出
可以看到他的正常启动，以及接受到源代理发送的数据
#在启动源第二个接事件的代理终端查看console输出
同样可以可以看到他的正常启动，以及接受到源代理发送的数据
Ok,成功
因为复用的流的事件要声明一个头部，然后我们检查头部对应的值，因为我们这边源类用http source
下面是源代理的配置
[html]view plaincopy
#配置文件：multi_source_case12.conf
a1.sources=r1
a1.sinks=k1k2
a1.channels=c1c2
#Describe/configurethesource
a1.sources.r1.type=org.apache.flume.source.http.HTTPSource
a1.sources.r1.port=50000
a1.sources.r1.host=192.168.233.128
a1.sources.r1.selector.type=multiplexing
a1.sources.r1.channels=c1c2
a1.sources.r1.selector.header=state
a1.sources.r1.selector.mapping.CZ=c1
a1.sources.r1.selector.mapping.US=c2
a1.sources.r1.selector.default=c1
#Describethesink
a1.sinks.k1.type=avro
a1.sinks.k1.channel=c1
a1.sinks.k1.hostname=192.168.233.129
a1.sinks.k1.port=50000
a1.sinks.k2.type=avro
a1.sinks.k2.channel=c2
a1.sinks.k2.hostname=192.168.233.130
a1.sinks.k2.port=50000
#Useachannelwhichbufferseventsinmemory
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.channels.c2.type=memory
a1.channels.c2.capacity=1000
a1.channels.c2.transactionCapacity=100
这里设置了2个channels与2个sinks 同时判断头部属性，当CZ的时，事件发送到sinks1,US时发送到sink2，其他的都发送到sink2，因此我们还有配置2个sinks对于的代理。这里的2个接受代理我们沿用之前复制的接受代理。
#敲命令
与之前复制的情况一样，首先先启动2个接受复制事件代理，如果先启动源发送的代理，会报他找不到sinks的绑定，因为2个接事件的代理还未起来。
flume-ng agent -cconf -f conf/multi_sink1_case12.conf -n a1-Dflume.root.logger=INFO,console
flume-ng agent -cconf -f conf/multi_sink2_case12.conf -n a1-Dflume.root.logger=INFO,console
在启动源发送的代理
flume-ng agent -cconf -f conf/multi_source_case12.conf -n a1-Dflume.root.logger=INFO,console
启动成功后
打开另一个终端输入，往侦听端口送数据
curl -X POST -d '[{"headers" :{"state" : "CZ"},"body" :"TEST1"}]' http://192.168.233.128:50000
curl -X POST -d '[{"headers" :{"state" : "US"},"body" :"TEST2"}]' http://192.168.233.128:50000
curl -X POST -d '[{"headers" :{"state" : "SH"},"body" :"TEST3"}]' http://192.168.233.128:50000
#在启动源发送的代理终端查看console输出
可以看到他的正常启动以及发送数据成功
#在启动源第一个接事件的代理终端查看console输出
这里可以清楚的看到，这个接事件代理只收到了2个事件，因为第二个事件因为我们设置复用，将头部信息对于的事件分流的关系，发送到另一个接事件代理去了。
#在启动源第二个接事件的代理终端查看console输出
Ok,第二个接事件代理因为复用分流，果然只获得了第二个事件信息。
声明：本站所有文章资源内容，如无特殊说明或标注，均为采集网络资源。如若本站内容侵犯了原著者的合法权益，可联系本站删除。
