flume对接kafka，模拟生产者实时生产数据

flume对接kafka，模拟生产者实时生产数据

引言

flume可以实时的监控日志，日志每增加一条，flume都会感知到，然后可以将这条新的数据传到kafka上，实际生产中，用户的每个行为，生成一条数据，存到日志或数据库中，然后使用flume来从日志中拉起数据。
任务：用shell脚本模拟用户行为每秒十条生成的数据存入到日志中，flume拉起日志中的数据传入kafka
已有数据：cmcc.json，目录日志：cmcc.log

1.编写脚本readcmcc.sh，从cmcc.json1秒10条追加到cmcc.log

for line in `cat /root/log/cmcc.json`
do
 	`echo $line >> /root/log/cmcc.log`
 	sleep 0.1s
done

2.编写flume脚本

agent.sources = s1                                                                                                                  
agent.channels = c1                                                                                                                 
agent.sinks = k1 
agent.sources.s1.type=exec
#监控的文件                                                                                            
agent.sources.s1.command=tail -F /root/log/cmcc.log                                                                           
agent.sources.s1.channels=c1                                                                                                        
agent.channels.c1.type=memory                                                                                                       
agent.channels.c1.capacity=10000                                                                                                    
agent.channels.c1.transactionCapacity=100 
#设置一个kafka接收器
agent.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink
#设置kafka的broker地址和端口号(所有的)
agent.sinks.k1.brokerList=hadoop01:9092,hadoop02:9092,hadoop03:9092
#设置kafka的topic
agent.sinks.k1.topic=cmcc2
#设置一个序列化方式
agent.sinks.k1.serializer.class=kafka.serializer.StringEncoder
#组装
agent.sinks.k1.channel=c1

3.启动kafka

 #启动kafka：
        nohup bin/kafka-server-start.sh  config/server.properties &
    #查看kafka的topic列表：
        bin/kafka-topics.sh --list --zookeeper hadoop02:2181
    #查看topic中的数据：
        bin/kafka-console-consumer.sh --zookeeper hadoop02:2181 --from-beginning --topic cmcc

4.执行flume脚本

bin/flume-ng agent -c conf -f conf/flume_kafka.sh -n agent -Dflume.root.logger=INFO,console

5.执行shell脚本

sh readcmcc.sh

6.kafka端查看

#查看有没有生成目标topic
bin/kafka-topics.sh --list --zookeeper hadoop02:2181
#读取次topic中的数据
bin/kafka-console-consumer.sh --zookeeper hadoop02:2181 --from-beginning --topic cmcckafka