前提是要先把flume和kafka独立的部分先搭建好。
下载flume-kafka-plus:https://github.com/beyondj2ee/flumeng-kafka-plugin
把lib目录下的
和package下的
都放到flume的lib目录
在插件包里有一个flume-conf.properties，把这个文件放到flume的conf文件夹里
然后修改以下内容
|
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 |
|
zkServer.sh start |
|
bin/kafka-server-start.sh config/server.properties |
|
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test |
|
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning |
|
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”就表示成功
看到前面我们下载的plugin版本不是最新的，是不是很不爽？不爽就换了他
我们用到的plugin的jar有：
像
我用的kafka版本是：kafka_2.11-0.8.2.1
替换的所换成这个版本的lib下的
kafka_2.11-0.8.2.1.jar
kafka-clients-0.8.2.1.jar
metrics-core-2.2.0.jar
metrics-annotation-2.2.0.jar照常用吧不用替换
scala-library-2.11.5.jar
scala-parser-combinators_2.11-1.0.2.jar
zkclient-0.3.jar
flumeng-kafka-plugin.jar这个要保持，他只是单纯的实现
替换完了再跑之前的测试就行
原文：http://www.cnblogs.com/gaopeng527/p/4968711.html
