进阶"/>
Flume进阶
案例1:双层Flume串联
双层flume衔接,第一层从exec采集sink到avro中,第二层从上一层的avro接收采集输出到控制台
第一层采用 exec source ===> memory channel ===> avro sink
第二层采用 avro source ===> memory channel ===> logger sink
Agent1
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444a1.sinks.k1.type = loggera1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Agent2
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/data/flumeData/word.log
a1.sources.r1.shell = /bin/bash -ca1.sinks.k1.type = avro
a1.sinks.k1.hostname = 0.0.0.0
a1.sinks.k1.port = 44444a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
启动脚本
bin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/exec-avro.conf \
--name a1 \
-Dflume.root.logger=INFO,consolebin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/avro-logger.conf \
--name a1 \
-Dflume.root.logger=INFO,console
案例2.双Source 单Sink
采用的是 exec 和 nc 两个source 使用 logger sink
Agent1
a1.sources = r1 r2
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/data/flumeData/word.log
a1.sources.r1.shell = /bin/bash -ca1.sources.r2.type = netcat
a1.sources.r2.bind = localhost
a1.sources.r2.port = 44444a1.sinks.k1.type = loggera1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sources.r2.channels = c1
a1.sinks.k1.channel = c1
启动脚本
bin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/twosource.conf \
--name a1 \
-Dflume.root.logger=INFO,console
3.单Soure双Sink
采用exec soure ,sink分别使用logger sink 和 hdfs sink
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/data/flumeData/word.log
a1.sources.r1.shell = /bin/bash -ca1.sinks.k1.type = loggera1.sinks.k2.type = hdfs
a1.sinks.k2.hdfs.path = hdfs://bigdata01:9000/flume/twosink
a1.sinks.k2.hdfs.filePrefix = bigdata-
a1.sinks.k2.hdfs.fileType = DataStream
a1.sinks.k2.hdfs.writeFormat = Texa1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1
启动脚本
bin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/twosink.conf \
--name a1 \
-Dflume.root.logger=INFO,console
Replacting Channel Selector(default)
不配置的话默认就是replacting Channel Selector,该Selector以复制的方式将event写到一至多个channel里面去。
nc soure ===> channel1 ===> sink1===> channel2 ===> sink2
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2a1.sources.r1.selector.type = replicatinga1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444a1.sinks.k1.type = loggera1.sinks.k2.type = hdfs
a1.sinks.k2.hdfs.path = hdfs://bigdata01:9000/flume/twosink
a1.sinks.k2.hdfs.filePrefix = bigdata-
a1.sinks.k2.hdfs.fileType = DataStream
a1.sinks.k2.hdfs.writeFormat = Texa1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2
启动脚本:
bin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/replactingchannelselector.conf \
--name a1 \
-Dflume.root.logger=INFO,console
Multiplexing Channel Selector
agent1 ===>
agent2 agent4 ===> multplexing ===>
agent3 ===>agent1 端口44441 ===> 55555 Inceptor 加US
agent2 端口44442 ===> 55555 Inceptor 加CN
agent3 端口44443 ===> 55555 Inceptor 加ENagent4 source 55555 ===> 根据header进行选择所发送到的channel
Agent1
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44441a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = state
a1.sources.r1.interceptors.i1.value = CNa1.sinks.k1.type = avro
a1.sinks.k1.hostname = 0.0.0.0
a1.sinks.k1.port = 55555a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Agent2
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44442a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = state
a1.sources.r1.interceptors.i1.value = USa1.sinks.k1.type = avro
a1.sinks.k1.hostname = 0.0.0.0
a1.sinks.k1.port = 55555a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Agent3
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44443a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = state
a1.sources.r1.interceptors.i1.value = ESa1.sinks.k1.type = avro
a1.sinks.k1.hostname = 0.0.0.0
a1.sinks.k1.port = 55555a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Agent4
a1.sources = r1
a1.sinks = k1 k2 k3
a1.channels = c1 c2 c3a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 55555a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = state
a1.sources.r1.selector.mapping.CN = c1
a1.sources.r1.selector.mapping.US = c2
a1.sources.r1.selector.default = c3a1.channels.c1.type = memory
a1.channels.c2.type = memory
a1.channels.c3.type = memorya1.sinks.k1.type = loggera1.sinks.k2.type = hdfs
a1.sinks.k2.hdfs.path = hdfs://bigdata01:9000/flume/sink2
a1.sinks.k2.hdfs.filePrefix = flume-
a1.sinks.k2.hdfs.fileType = DataStream
a1.sinks.k2.hdfs.writeFormat = Texa1.sinks.k3.type = hdfs
a1.sinks.k3.hdfs.path = hdfs://bigdata01:9000/flume/sink3
a1.sinks.k3.hdfs.filePrefix = flume-
a1.sinks.k3.hdfs.fileType = DataStream
a1.sinks.k3.hdfs.writeFormat = Texa1.sources.r1.channels = c1 c2 c3
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2
a1.sinks.k3.channel = c3
启动脚本
bin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/multi1.conf \
--name a1 \
-Dflume.root.logger=INFO,consolebin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/multi2.conf \
--name a1 \
-Dflume.root.logger=INFO,consolebin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/multi3.conf \
--name a1 \
-Dflume.root.logger=INFO,consolebin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/multi4.conf \
--name a1 \
-Dflume.root.logger=INFO,console
Flume Sink Processors
三种类型分别是fail over、load_balance、default,生产上面采用fail over这种方式。
avro sink----avro source ===> logger sink
nc source ===> memory channel ===>avro sink----avro source ===> logger sink
数值越大,优先级越高。测试先启动两个avro-logger1,发送数据,发现数据都往55552中去了,手动停掉55552,再次发送数据发现 Agent3报错,随后又重新将数据发送到55551上面,测试failover功能成功。
Agent1
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 55551a1.sinks.k1.type = loggera1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Agent2
a1.sources = r1
a1.sinks = k1
a1.channels = c1a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 55552a1.sinks.k1.type = loggera1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Agent3
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
a1.sinkgroups.g1.processor.maxpenalty = 10000a1.channels.c1.type = memorya1.sinks.k1.type = avro
a1.sinks.k1.hostname = 0.0.0.0
a1.sinks.k1.port = 55551a1.sinks.k2.type = avro
a1.sinks.k2.hostname = 0.0.0.0
a1.sinks.k2.port = 55552a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1
启动脚本
bin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/sinkprocessors/avro-logger1.conf \
--name a1 \
-Dflume.root.logger=INFO,consolebin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/sinkprocessors/avro-logger2.conf \
--name a1 \
-Dflume.root.logger=INFO,consolebin/flume-ng agent \
--conf /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/conf \
--conf-file /home/hadoop/app/flume-1.6.0-cdh5.16.2-bin/script/sinkprocessors/nc-multi-sink.conf \
--name a1 \
-Dflume.root.logger=INFO,console
更多推荐
Flume进阶
发布评论