flume消费kafka到hdfs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# source alias
agent.sources = source_from_kafka
# channels alias
agent.channels = mem_channel
# sink alias
agent.sinks = hdfs_sink
# define kafka source
agent.sources.source_from_kafka.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.source_from_kafka.channels = mem_channel
agent.sources.source_from_kafka.batchSize = 5000
# set kafka broker address
agent.sources.source_from_kafka.kafka.bootstrap.servers = master:9092,slave1:9092,slave2:9092
# set kafka topic
agent.sources.source_from_kafka.kafka.topics = test
# set kafka groupid
agent.sources.source_from_kafka.kafka.consumer.group.id = test-consumer-group
# defind hdfs sink
agent.sinks.hdfs_sink.type = hdfs
# specify the channel the sink should use
agent.sinks.hdfs_sink.channel = mem_channel
# set store hdfs path
agent.sinks.hdfs_sink.hdfs.path = /data/flume/kafka/%Y%m%d
# set file size to trigger roll
agent.sinks.hdfs_sink.hdfs.rollSize = 0
agent.sinks.hdfs_sink.hdfs.rollCount = 0
agent.sinks.hdfs_sink.hdfs.rollInterval = 3600
agent.sinks.hdfs_sink.hdfs.threadsPoolSize = 30
agent.sinks.hdfs_sink.hdfs.fileType=DataStream
agent.sinks.hdfs_sink.hdfs.writeFormat=Text
# define channel from kafka source to hdfs sink
agent.channels.mem_channel.type = memory
# channel store size
agent.channels.mem_channel.capacity = 100000
# transaction size
agent.channels.mem_channel.transactionCapacity = 10000

1
2
./flume-ng agent --conf ../conf/ --name agent --conf-file ../conf/flume-hdfs.conf -Dflume.root.logger=INFO,console

1
2
3
./kafka-server-start.sh
-daemon ../config/server.properties

1
2
./kafka-console-producer.sh --broker-list 192.168.88.5:9092 --topic test

1
2
Hadoop fs -cat /data/flume/kafka/20190620/FlumeData.1561088964259.tmp