Centos7.4环境下flume集成kafka

一.练习目标

1.Flume.监控/tmp/logs/update.log文件

2.创建日志生成的脚本，写入/tmp/logs/update.log。

3.Flume捕获数据

4.将捕获的数据，推送到消息队列kafka

5.在Kafka模拟数据消费 

二.详细步骤

1.解压并安装flume

解压到指定目录:
tar -zxvf apache-flume-1.6.0-bin.tar.gz -C /opt/soft/
添加环境变量：
vim /etc/profile

更新环境变量：
source /etc/profile

2.flume集成kafka配置文件

在/usr/local/flume/apache-flume-1.6.0-bin/conf下，创建flume_kafka.conf。（我的Flume的安装路径是这样的）

a1.sources = r1
​
a1.channels = c1
​
a1.sinks = k1
​
a1.sources.r1.type = exec
​
#tail -F 根据文件名进行追踪
​
a1.sources.r1.command = tail -F /tmp/logs/update.log
​
#把source和channel关联在一起
a1.sources.r1.channels = c1
​​
a1.channels.c1.type=memory
​
a1.channels.c1.capacity=10000
​
a1.channels.c1.transactionCapacity=100
​​
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
​
#指定kafka类型
​
a1.sinks.k1.topic = update_log
​
#kafka集群地址
​
a1.sinks.k1.brokerList = Master:9092,Slave1:9092,Slave2:9092
​
a1.sinks.k1.requiredAcks = 1
​
a1.sinks.k1.batchSize = 20
​
a1.sinks.k1.channel = c1

3.准备数据

1)创建/tmp/logs/update.log文件

[root@Master ~]# mkdir /tmp/logs
[root@Master ~]# cd /tmp/logs
[root@Master logs]# touch update.log
2)创建往update.log日志文件添加数据脚本 create_log.sh 文件
[root@Master tmp]# pwd
/tmp
[root@Master tmp]# vi  create_log.sh
create_log.sh 内容如下：
​
#!/bin/bash
​
while true
​
do
​
echo "ranyi_love_guanyuhang" >> /tmp/logs/update.log;
​
sleep 0.5;
​
done
 
3）创建update_log的topic

kafka-topics.sh --create --zookeeper Master:2181  --partitions 3 --replication-factor 2 --topic update_log

4.启动

1）进入flume安装目录下启动flume。（kafka.conf是flume集成kafka的配置内容）
flume-ng agent -c conf -f conf/flume_kafka.conf -n a1  -Dflume.root.logger=INFO,console
 
2）启动模拟生产日志数据脚本
[root@Master tmp]# sh create_log.sh
 
3）在kafka安装的bin目录下启动kafka消费者查看数据
kafka-console-consumer.sh --from-beginning  --zookeeper Master:2181 --topic update_log