1.Flume.监控/tmp/logs/update.log文件
2.创建日志生成的脚本，写入/tmp/logs/update.log。
3.Flume捕获数据
4.将捕获的数据，推送到消息队列kafka
5.在Kafka模拟数据消费
tar -zxvf apache-flume-1.6.0-bin.tar.gz
vim /etc/profile
添加进环境变量
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
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
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
可以看到数据不断增加
原文：https://www.cnblogs.com/phy2020/p/12801253.html