Flume 整合 Kafka

一、背景

先说一下，为什么要使用 Flume + Kafka？

以实时流处理项目为例，由于采集的数据量可能存在峰值和峰谷，假设是一个电商项目，那么峰值通常出现在秒杀时，这时如果直接将 Flume 聚合后的数据输入到 Storm 等分布式计算框架中，可能就会超过集群的处理能力，这时采用 Kafka 就可以起到削峰的作用。Kafka 天生为大数据场景而设计，具有高吞吐的特性，能很好地抗住峰值数据的冲击。

二、整合流程

Flume 发送数据到 Kafka 上主要是通过 KafkaSink 来实现的，主要步骤如下：

1. 启动Zookeeper和Kafka

这里启动一个单节点的 Kafka 作为测试：

1
2
3
4
5
6
7
1# 启动Zookeeper

2zkServer.sh start

3

4# 启动kafka

5bin/kafka-server-start.sh config/server.properties

6

7

2. 创建主题

创建一个主题 flume-kafka，之后 Flume 收集到的数据都会发到这个主题上：

1
2
3
4
5
6
7
8
9
10
1# 创建主题

2bin/kafka-topics.sh --create \

3--zookeeper luya03:2181 \

4--replication-factor 1   \

5--partitions 1 --topic flume-kafka

6

7# 查看创建的主题

8bin/kafka-topics.sh --zookeeper luya03:2181 --list

9

10

3. 启动kafka消费者

启动一个消费者，监听我们刚才创建的 flume-kafka 主题：

1
2
3
1# bin/kafka-console-consumer.sh --bootstrap-server luya03:9092 --topic flume-kafka

2

3

4. 配置Flume

新建配置文件 exec-memory-kafka.properties，文件内容如下。这里我们监听一个名为 kafka.log 的文件，当文件内容有变化时，将新增加的内容发送到 Kafka 的 flume-kafka 主题上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
1a1.sources = s1

2a1.channels = c1

3a1.sinks = k1                                                                                         

4

5a1.sources.s1.type=exec

6a1.sources.s1.command=tail -F /tmp/kafka.log

7a1.sources.s1.channels=c1 

8

9#设置Kafka接收器

10a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink

11#设置Kafka地址

12a1.sinks.k1.brokerList=luya03:9092

13#设置发送到Kafka上的主题

14a1.sinks.k1.topic=flume-kafka

15#设置序列化方式

16a1.sinks.k1.serializer.class=kafka.serializer.StringEncoder

17a1.sinks.k1.channel=c1     

18

19a1.channels.c1.type=memory

20a1.channels.c1.capacity=10000

21a1.channels.c1.transactionCapacity=100   

22

23

5. 启动Flume

1
2
3
4
5
6
1flume-ng agent \

2--conf conf \

3--conf-file /usr/app/apache-flume-1.6.0-cdh5.15.2-bin/examples/exec-memory-kafka.properties \

4--name a1 -Dflume.root.logger=INFO,console

5

6

6. 测试

向监听的 /tmp/kafka.log 文件中追加内容，查看 Kafka 消费者的输出：

可以看到 flume-kafka 主题的消费端已经收到了对应的消息：