blog/BigData-Notes
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Flume 整合 Kafka

Browse Source
This commit is contained in:
罗祥 2019-06-03 10:31:57 +08:00
parent 88f43dd809
commit b96c1264c4
1 changed files with 105 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
notes/Flume整合Kafka.md
View File

@ -1,5 +1,6 @@
# Flume 整合 Kafka
<nav>
<nav>
<a href="#一背景">一、背景</a><br/>
<a href="#二整合流程">二、整合流程</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#1-启动Zookeeper和Kafka">1. 启动Zookeeper和Kafka</a><br/>
@ -14,7 +15,7 @@
先说一下，为什么要使用 Flume + Kafka？
这里举一个实时流处理的项目为例，由于采集的日志数据可能存在峰值和峰谷，比如如果是一个电商项目，那么峰值就会出现在秒杀时，这时如果直接将Flume聚合后的数据输入到Storm或者Spark Streaming 中进行处理，集群处理压力就会过大，这时采用Kafka就可以起到削峰的作用。Kafka天生就是为大数据场景而设计，具有高吞吐等特性，能很好的抗住峰值数据的冲击。
以实时流处理项目为例，由于采集的数据量可能存在峰值和峰谷，假设是一个电商项目，那么峰值通常出现在秒杀时，这时如果直接将Flume聚合后的数据输入到Storm等分布式计算框架中，可能就会超过集群的处理能力，这时采用Kafka就可以起到削峰的作用。Kafka天生为大数据场景而设计，具有高吞吐的特性，能很好地抗住峰值数据的冲击。
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/flume-kafka.png"/> </div>
@ -24,9 +25,9 @@
Flume发送数据到Kafka上主要是通过`KafkaSink`来实现的，主要步骤如下：
#### 1. 启动Zookeeper和Kafka
### 1. 启动Zookeeper和Kafka
这里可以只启动一个单节点的Kafka作为测试
这里启动一个单节点的Kafka作为测试：
```shell
# 启动Zookeeper
@ -36,9 +37,9 @@ zkServer.sh start
bin/kafka-server-start.sh config/server.properties
```
#### 2. 创建主题
### 2. 创建主题
创建一个主题`flume-kafka`，之后Flume收集到的数据都会发到这个主题上
创建一个主题`flume-kafka`，之后Flume收集到的数据都会发到这个主题上：
```shell
# 创建主题
@ -53,9 +54,9 @@ bin/kafka-topics.sh --zookeeper hadoop001:2181 --list
#### 3. 启动kafka消费者
### 3. 启动kafka消费者
启动一个消费者，监听我们刚才创建的`flume-kafka`主题
启动一个消费者，监听我们刚才创建的`flume-kafka`主题：
```shell
# bin/kafka-console-consumer.sh --bootstrap-server hadoop001:9092 --topic flume-kafka
@ -63,9 +64,9 @@ bin/kafka-topics.sh --zookeeper hadoop001:2181 --list
#### 4. 配置Flume
### 4. 配置Flume
新建配置文件`exec-memory-kafka.properties`，文件内容如下。这里我们监听一个名为`kafka.log`的文件，当文件内容有变化时，将新增加的内容发送到Kafka的`flume-kafka`主题中。
新建配置文件`exec-memory-kafka.properties`，文件内容如下。这里我们监听一个名为`kafka.log`的文件，当文件内容有变化时，将新增加的内容发送到Kafka的`flume-kafka`主题上。
```properties
a1.sources = s1
@ -93,7 +94,7 @@ a1.channels.c1.transactionCapacity=100
#### 5. 启动Flume
### 5. 启动Flume
```shell
flume-ng agent \
@ -104,12 +105,12 @@ flume-ng agent \
#### 6. 测试
### 6. 测试
向监听的`/tmp/kafka.log `文件中追加内容，查看Kafka消费者的输出
向监听的`/tmp/kafka.log `文件中追加内容，查看Kafka消费者的输出：
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/flume-kafka-01.png"/> </div>
可以看到`flume-kafka`主题的消费端已经收到了对应的消息
可以看到`flume-kafka`主题的消费端已经收到了对应的消息：
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/flume-kafka-2.png"/> </div>