flume实现kafka到hdfs实时数据采集 - 有负载均衡策略

方案:
两台采集机器dc007.dx.momo.com,dc008.dx.momo.com.有两个写hdfs的sink,分别部署到两台机器,然后两个负载均衡的agent,也是分布部属到两台机器上,每一个agent都是写到两个hdfs sink的source端.

配置:

*******************************************hdfs sink

hdfs-sink.sources = r1

hdfs-sink.sinks = k1

hdfs-sink.channels = c1

# Describe/configure the source

hdfs-sink.sources.r1.type = avro

hdfs-sink.sources.r1.channels = c1

hdfs-sink.sources.r1.bind = dc008.dx.momo.com

hdfs-sink.sources.r1.port = 5555

# Describe the sink

hdfs-sink.sinks.k1.type = hdfs

#保证每天的每个小时一个文件夹

hdfs-sink.sinks.k1.hdfs.path = hdfs://nameservice1/user/dc/test/flume/sdk_function_log/%Y%m%d/%Y%m%d%H

hdfs-sink.sinks.k1.hdfs.filePrefix = base.log

#如果n秒没有写文件就自动关闭hdfs文件,当每个小时结束的时候可以在10s后关闭文件

hdfs-sink.sinks.k1.hdfs.idleTimeout = 10

#以文本形式写入hdfs

hdfs-sink.sinks.k1.hdfs.fileType=DataStream

hdfs-sink.sinks.k1.hdfs.writeFormat=Text

#控制文件大小

hdfs-sink.sinks.k1.hdfs.rollInterval=0

#256mb

hdfs-sink.sinks.k1.hdfs.rollSize=256000000

hdfs-sink.sinks.k1.hdfs.rollCount=0

# Use a channel which buffers events in memory

hdfs-sink.channels.c1.type = memory

hdfs-sink.channels.c1.capacity = 10000

hdfs-sink.channels.c1.transactionCapacity = 10000

# Bind the source and sink to the channel

hdfs-sink.sources.r1.channels = c1

hdfs-sink.sinks.k1.channel = c1

*******************************************

*******************************************load balance

lb-kafka-hdfs.sources=r1

lb-kafka-hdfs.sinks=k1 k2

lb-kafka-hdfs.channels=c1

#failover conf

lb-kafka-hdfs.sinkgroups = g1

lb-kafka-hdfs.sinkgroups.g1.sinks = k1 k2

lb-kafka-hdfs.sinkgroups.g1.processor.type = load_balance

lb-kafka-hdfs.sinkgroups.g1.processor.backoff = true

lb-kafka-hdfs.sinkgroups.g1.processor.selector = round_robin

#source conf

lb-kafka-hdfs.sources.r1.type = org.apache.flume.source.kafka.KafkaSource

lb-kafka-hdfs.sources.r1.channels = c1

lb-kafka-hdfs.sources.r1.zookeeperConnect = dc002.dx:2181,dc003.dx:2181,dc004.dx:2181,dc005.dx:2181,dc006.dx:2181/kafka_0.8.2.2

lb-kafka-hdfs.sources.r1.groupId = flume-test

lb-kafka-hdfs.sources.r1.topic = sdk_function_log

lb-kafka-hdfs.sources.r1.kafka.consumer.timeout.ms = 100

#sink conf

lb-kafka-hdfs.sinks.k1.type = avro

lb-kafka-hdfs.sinks.k1.channel = c1

lb-kafka-hdfs.sinks.k1.hostname = dc007.dx.momo.com

lb-kafka-hdfs.sinks.k1.port = 5555

lb-kafka-hdfs.sinks.k2.type = avro

lb-kafka-hdfs.sinks.k2.channel = c1

lb-kafka-hdfs.sinks.k2.hostname = dc008.dx.momo.com

lb-kafka-hdfs.sinks.k2.port = 5555

#channel conf

lb-kafka-hdfs.channels.c1.type = memory

lb-kafka-hdfs.channels.c1.capacity = 10000

lb-kafka-hdfs.channels.c1.transactionCapacity = 10000

*******************************************

启动命令:

1.先启动两个sink,在dc007.dx,dc008.dx

flume-ng agent --conf /home/dc/datacenter/soft/flume/default/conf -f /home/dc/datacenter/src/flume-conf/lb_kafka_hdfs/hdfs-sink.conf -Dflume.root.logger=INFO,console -n hdfs-sink

2.启动两个负载服务,在dc007.dx,dc008.dx

flume-ng agent --conf /home/dc/datacenter/soft/flume/default/conf -f /home/dc/datacenter/src/flume-conf/lb_kafka_hdfs/lb-kafka-hdfs.conf -Dflume.root.logger=INFO,console -n lb-kafka-hdfs