Flume抽取Kafka数据到HDFS

Flume抽取Kafka数据到HDFS

1）数据抽取

在/export/servers/apache-flume-1.8.0-bin/conf下创建kafka-hdfs.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
​
# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.kafka.bootstrap.servers = node01:9092,node02:9092,node03:9092
a1.sources.r1.kafka.consumer.group.id = flume_test
a1.sources.r1.kafka.topics = test2
​
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://node01:8020/rawlog/20%y-%m-%d/%H
a1.sinks.k1.hdfs.useLocalTimeStamp = false
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.filePrefix = test
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize =  12800000000
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.threadsPoolSize = 10
a1.sinks.k1.hdfs.batchSize = 2000
a1.sinks.k1.hdfs.threadsPoolSize = 10
​
​
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1500000
a1.channels.c1.transactionCapacity = 10000
​
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
​

2）下载Hadoop依赖

从Hadoop集群中获取相关jar包放到flume/lib文件夹下

hadoop-common-2.6.3.jar hadoop-hdfs-2.6.3.jar commons-configuration-1.6.jar hadoop-auth-2.6.3.jar htrace-core-3.0.4.jar

3）启动

在node01上启动

nohup bin/flume-ng agent -n a1 -c conf -f conf/kafka-hdfs.conf >/dev/null 2>&1 &

bin/flume-ng agent -c conf -f conf/kafka-hdfs.conf -n a1 -Dflume.root.logger=INFO,console