Flume+Kafka联通

前提

前提是要先把flume和kafka独立的部分先搭建好。
下载插件包
下载flume-kafka-plus:https://github.com/beyondj2ee/flumeng-kafka-plugin
把lib目录下的

和package下的

都放到flume的lib目录
修改原有的flume-conf文件
在插件包里有一个flume-conf.properties，把这个文件放到flume的conf文件夹里

修改以下内容

1
2
3
4
5
6
7

producer.sources.s.type = exec
producer.sources.s.command = tail -f -n+1 ~/tmp/test.log
producer.sources.s.channels = c
……
producer.sinks.r.custom.topic.name=test
……
consumer.sources.s.custom.topic.name=test

producer.sources.s.type = exec
producer.sources.s.command = tail -f -n+1 ~/tmp/test.log
producer.sources.s.channels = c

以上路径处需要注意：

• 尽量不要使用/tmp/这种目录格式，代表的是当前用户的目录。
• 使用tail -f filename 就可以了这里
• 文件目录要对应正确路径，且对应位置要有相应文件！！！！
• 与文件权限无关

启动zookeeper

zkServer.sh start

启动kafka broker

bin/kafka-server-start.sh config/server.properties

创建kafka topic

bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic test

启动kafka consumer

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –topic test –from-beginning

启动flume

bin/flume-ng agent –conf conf –conf-file conf/flume-conf.properties –name producer -Dflume.root.logger=INFO,console

测试

echo “this is a test” >> ~/tmp/test.txt
此时只要能在consumer里现“this is a test”就表示成功

测试成功

<< 更多精彩尽在『程序萌部落』>>
<< https://www.cxmoe.com >>