1.进入flume官网 www.org.apache.flume.com
点击第三个
2.使用kafka前需要先开启 zookeeper zkServer.sh start
启动flume flume-ng version //这个好像不影响 保险起见还是启动了
启动kafka kafka-server-start.sh /opt/soft/kafka200/config/server.properties
3.编写配置文件
读文件首先最好去头 写好配置文件
#user是agent的名称，user中定义了一个叫r1的source，如果有多个，使用空格间隔 user.sources = r1 user.sinks = k1 user.channels = c1 #组名名.属性名=属性值 #定义source user.sources.r1.type=spooldir user.sources.r1.spoolDir=/opt/mydata/users/ user.sources.r1.interceptors=i1 user.sources.r1.interceptors.i1.type = regex_filter user.sources.r1.interceptors.i1.regex=user.* user.sources.r1.interceptors.i1.excludeEvents=true #设置读写行数 user.sources.r1.deserializer.maxLineLength=204800 #定义chanel user.channels.c1.type=memory #定义sink user.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink user.sinks.k1.kafka.topic=users user.sinks.k1.kafka.bootstrap.servers=192.168.80.181:9092 #连接组件 同一个source可以对接多个channel，一个sink只能从一个channel拿数据！ user.sources.r1.channels=c1 user.sinks.k1.channel=c1
文件名称编写为users.conf 放在这个/opt/fconf 目录下
4.查看kafka分区
kafka-topics.sh --zookeeper 192.168.80.181:2181 --list
5.删除分区 这里
kafka-topics.sh --zookeeper 192.168.80.181:2181 --delete --topic mydemo03
6.flume 读取·文件
flume-ng agent --name user -f /opt/fconf/users.conf
7.开启消费者监听
kafka-console-consumer.sh --bootstrap-server 192.168.80.181:9092 --topic users --from-beginning
8.查看分期区数据
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.80.181:9092 --topic users --time -1
wc -l 你的文件 可以查看文件有多少行数据 用来对比数据
