一.练习目标
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
vim /etc/profile
添加进环境变量
2.flume集成kafka配置文件
cd conf/ vim flume_kafka.conf 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 = hdp01:9092,hdp02:9092,hdp03:9092 a1.sinks.k1.requiredAcks = 1 a1.sinks.k1.batchSize = 20 a1.sinks.k1.channel = c1
2.准备数据
1)创建/tmp/logs/update.log文件
cd /tmp/logs vim update.log
2)创建往update.log日志文件添加数据脚本 create_log.sh 文件
#!/bin/bash
while true
do
echo "ranyi_love_guanyuhang" >> /tmp/logs/update.log;
time3=$(date "+%Y-%m-%d %H:%M:%S")
echo $time3 >> /tmp/logs/update.log;
sleep 0.5;
done
3)创建update_log的topic
3.启动
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集群中启动消费者消费数据
kafka-console-consumer.sh --from-beginning --zookeeper Master:2181 --topic update_log
可以看到数据不断增加