Flume进阶

进阶"/>

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