Flume1.6+Kafka

需求：

　　使用Flume读取Linux的文本文件，将文本文件中的数据发送到kafka

前置条件：

　　安装好zk集群，安装好kafka集群

配置Flume

1.在Flume官方下载最新的flume

　　wget   http://mirrors.shu.edu.cn/apache/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz

2.解决flume安装包

 cd /export/software/

tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /export/servers/

cd /export/servers/

ln -s apache-flume-1.9.0-bin flume

su -root

vi /etc/profile

export FLUME_HOME=/home/hadoop/export/severs/flume

export PATH=$PATH:$FLUME_HOME/bin

3.创建flume配置文件

cd /export/servers/flume/conf/

mkdir myconf

vi exec.conf

输入以下内容：

a1.sources = r1
a1.channels = c1
a1.sinks = k1

a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/export/data/flume_sources/click_log/1.log
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
a1.sinks.k1.topic = orderMq 
a1.sinks.k1.brokerList = min5:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20
a1.sinks.k1.channel = c1

a1.sinks.k1.kafka.bootstrap.servers = min5:9092,min6:9092,min7:9092

　　注：配置完毕，flume环节的工作基本完成，接下来准备目标数据文件。

4.准备目标数据的目录

mkdir -p /home/hadoop/export/data/flume_sources/click_log

5.通过脚本创建目标文件并生产数据

for((i=0;i<=50000;i++));
do echo "message-"+$i >>/home/hadoop/export/data/flume_sources/click_log/1.log;
done

注意：脚本名称叫做click_log_out.sh需要使用root用户赋权.chmod +x click_log_out.sh

6.开始打通所有流程

第一步：启动kafka集群

　cd 到kafka/bin下　

nohup ./kafka-server-start.sh /home/hadoop/export/servers/kafka/config/server.properties &

　　

第二步：创建一个topic并开启consumer

kafka-console-consumer.sh --topic=test --zookeeper min5:2181

　　

第三步：执行产生数据的脚本

sh click_log_out.sh

　　

第四步：启动flume客户端

./bin/flume-ng agent -n a1 -c conf -f conf/myconf/exec.conf -Dflume.root.logger=INFO,console

　

 

第五步：在第三步启动的kafka consumer 窗口查看效果