08_Flume_Selector实践

 

 

实践一：replicating selector

1、目标场景

selector将event复制，分发给所有下游节点

2、Flume Agent配置

Agent配置

# Name the components on this agent  
a1.sources = r1  
a1.sinks = k1 k2  
a1.channels = c1 c2  
   
# http source, with replicating selector 
a1.sources.r1.type = http
a1.sources.r1.port = 6666 
a1.sources.r1.bind = master
a1.sources.r1.selector.type = replicating  

# Describe the sink  
a1.sinks.k1.type = avro  
a1.sinks.k1.hostname = slave1  # bind to remote host,RPC
a1.sinks.k1.port = 6666


a1.sinks.k2.type = avro  
a1.sinks.k2.hostname = slave2  # bind to remote host,PRC
a1.sinks.k2.port = 6666

# 2 channels in selector test 
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  

# bind source ,sink to channels
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1  
a1.sinks.k2.channel = c2  

Collector1配置

# 01 specify agent,source,sink,channel
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 02 avro source,connect to local port 6666
a1.sources.r1.type = avro
a1.sources.r1.bind = slave1
a1.sources.r1.port = 6666

# 03 logger sink
a1.sinks.k1.type = logger 

# 04 channel,memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 05 bind source,sink to channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Collector2配置

# 01 specify agent,source,sink,channel
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 02 avro source,connect to local port 6666
a1.sources.r1.type = avro
a1.sources.r1.bind = slave2
a1.sources.r1.port = 6666

# 03 logger sink
a1.sinks.k1.type = logger 

# 04 channel,memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 05 bind source,sink to channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

3、验证Replicating selector

Agent端通过curl -X POST -d 'json数据'  模拟HTTP请求，Agent Souce将其转换为event，并复制两份，分别发给Collector1， Collector2

* 模拟的HTTP请求

* Collector1收到的event

* Collector2收到的event

4、replicating selector的官网配置参考

实践二：multiplexing selector

 1、目标场景

2、Flume Agent配置

Agent配置

# Name the components on this agent  
a1.sources = r1  
a1.sinks = k1 k2  
a1.channels = c1 c2  
   
# http source,with multiplexing selector 
a1.sources.r1.type = http
a1.sources.r1.bind = master
a1.sources.r1.port = 6666  

# for header with key country
# send to c1 if country's value is china
# send to c2 if country's value is singapore
a1.sources.r1.selector.type= multiplexing  
a1.sources.r1.selector.header= country
a1.sources.r1.selector.mapping.china = c1  
a1.sources.r1.selector.mapping.singapore = c2  
a1.sources.r1.selector.default= c1 

# Describe the sink  
a1.sinks.k1.type = avro  
a1.sinks.k1.hostname = slave1   # bind to remote host, RPC
a1.sinks.k1.port = 6666
       
a1.sinks.k2.type = avro  
a1.sinks.k2.hostname = slave2   # bind to remote host, RPC
a1.sinks.k2.port = 6666

# 2 channels, for selection
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  

# bind source,sink to channels
a1.sources.r1.channels= c1 c2  
a1.sinks.k1.channel = c1  
a1.sinks.k2.channel = c2  

Collector1配置

# 01 specify agent,source,sink,channel
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 02 avro source,connect to local port 6666
a1.sources.r1.type = avro
a1.sources.r1.bind = slave1
a1.sources.r1.port = 6666

# 03 logger sink
a1.sinks.k1.type = logger 

# 04 channel,memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 05 bind source,sink to channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Collector2配置

# 01 specify agent,source,sink,channel
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 02 avro source,connect to local port 6666
a1.sources.r1.type = avro
a1.sources.r1.bind = slave2
a1.sources.r1.port = 6666

# 03 logger sink
a1.sinks.k1.type = logger 

# 04 channel,memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 05 bind source,sink to channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

3、验证multiplexing selector

验证思路：

Agent端通过curl -X POST -d 'json数据' 模拟HTTP请求，Agent Souce将其转换为event，并根据header中key为country的不同value值，进行分发
value为china,则分发给C1，最终到达Collecotor1; value为singapore, 则分发给C2，最终到达Collector2; 其他取值，则分发给默认通道C1

1）发送带有country:china的HTTP请求

2）Collecotor1收到并在终端打印出flume event

3）发送带有country:singapore的HTTP请求

4) Collector2收到并在终端打印出flume event

5) 发送带有country:unknown的HTTP请求

6) Collector1因为被配置为默认通道，因此收到该flume event，并打印到终端

4、multiplexing selector官方配置参考