實踐一：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官方配置參考
