Kafka和Flume整合

Kafka与flume整合流程

Kafka整合flume流程图
flume主要是做日志数据(离线或实时)地采集。

图-1 数据处理

图-1显示的是flume采集完毕数据之后，进行的离线处理和实时处理两条业务线，现在再来学习flume和kafka的整合处理。
配置flume.conf文件
配置如下：

//为我们的source channel sink起名
a1.sources = r1
a1.channels = c1
a1.sinks = k1
//指定我们的source收集到的数据发送到哪个管道 a1.sources.r1.channels = c1
//指定我们的source数据收集策略
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /opt/module/flumedata
a1.sources.r1.deletePolicy = never
a1.sources.r1.fileSuffix =.COMPLETED
a1.sources.r1.ignorePattern = ^(.)*\.tmp$
a1.sources.r1.inputCharset = UTF-8
//指定我们的channel为memory,即表示所有的数据都装进memory当中 a1.channels.c1.type = memory
//指定我们的sink为kafka sink，并指定我们的sink从哪个channel当中读取数据
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = hadoop
a1.sinks.k1.kafka.bootstrap.servers=hadoop101:9092,hadoop102:9092,hadoop103:9092
a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks
= 1

启动flume和kafka的整合测试
1)消费者监听读取的数据。

[root@hadoop102 kafka]$ bin/kafka-console-consumer.sh
–topic test
–bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092
–from-beginning

2)启动flume-agent。

[root@hadoop101 flume]$ bin/flume-ng agent --conf conf --conf-file
conf/flume_kafka.conf --name a1 -Dflume.root.logger=INFO,console

3)发送与接收数据验证。
4)验证结果：显示的发送与接收数据，可以说明flume和kafka的整合成功。