flume實作kafka到hdfs實時資料采集 - 有負載均衡政策

方案:

兩台采集機器pc1,pc2.有兩個寫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 = pc2

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 = pc002:2181,pc003:2181,pc004:2181,pc005:2181,pc006: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 = pc1

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 = pc2

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,在pc1,pc2

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.啟動兩個負載服務,在pc1,pc2

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