使用flume+kafka+storm构建实时日志分析系统
本文只会涉及flume和kafka的结合，kafka和storm的结合可以参考其他博客1. flume安装使用
下载flume安装包http://www.apache.org/dyn/closer.cgi/flume/1.5.2/apache-flume-1.5.2-bin.tar.gz
解压$ tar -xzvf apache-flume-1.5.2-bin.tar.gz -c /opt/flume
flume配置文件放在conf文件目录下，执行文件放在bin文件目录下。
1）配置flume
进入conf目录将flume-conf.properties.template拷贝一份，并命名为自己需要的名字
$ cp flume-conf.properties.template flume.conf
修改flume.conf的内容，我们使用file sink来接收channel中的数据，channel采用memory channel，source采用exec source，配置文件如下：
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>agent.sources = seqgensrc</li><li>agent.channels = memorychannel</li><li>agent.sinks = loggersink</li><li></li><li># for each one of the sources, the type is defined</li><li>agent.sources.seqgensrc.type = exec</li><li>agent.sources.seqgensrc.command = tail -f /data/mongodata/mongo.log</li><li>#agent.sources.seqgensrc.bind = 172.168.49.130</li><li></li><li># the channel can be defined as follows.</li><li>agent.sources.seqgensrc.channels = memorychannel</li><li></li><li># each sink's type must be defined</li><li>agent.sinks.loggersink.type = file_roll</li><li>agent.sinks.loggersink.sink.directory = /data/flume</li><li></li><li>#specify the channel the sink should use</li><li>agent.sinks.loggersink.channel = memorychannel</li><li></li><li># each channel's type is defined.</li><li>agent.channels.memorychannel.type = memory</li><li></li><li># other config values specific to each type of channel(sink or source)</li><li># can be defined as well</li><li># in this case, it specifies the capacity of the memory channel</li><li>agent.channels.memorychannel.capacity = 1000</li><li>agent.channels.memory4log.transactioncapacity = 100</li></ol>2）运行flume agent
切换到bin目录下，运行一下命令：
$ ./flume-ng agent --conf ../conf -f ../conf/flume.conf --n agent -dflume.root.logger=info,console
在/data/flume目录下可以看到生成的日志文件。
2. 结合kafka
由于flume1.5.2没有kafka sink，所以需要自己开发kafka sink
可以参考flume 1.6里面的kafka sink，但是要注意使用的kafka版本，由于有些kafka api不兼容的
这里只提供核心代码，process()内容。
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>sink.status status = status.ready;<br /> </li><li><br /></li><li>channel ch = getchannel();<br /></li><li>transaction transaction = null;<br /></li><li>event event = null;<br /></li><li>string eventtopic = null;<br /></li><li>string eventkey = null;<br /></li><li><br /></li><li>try {<br /></li><li>transaction = ch.gettransaction();<br /></li><li>transaction.begin();<br /></li><li>messagelist.clear();<br /></li><li><br /></li><li>if (type.equals("sync")) {<br /></li><li>event = ch.take();<br /></li><li><br /></li><li> if (event != null) {<br /></li><li> byte[] tempbody = event.getbody();<br /></li><li> string eventbody = new string(tempbody,"utf-8");<br /></li><li> map<string, string> headers = event.getheaders();<br /></li><li><br /></li><li> if ((eventtopic = headers.get(topic_hdr)) == null) {<br /></li><li> eventtopic = topic;<br /></li><li> }<br /></li><li><br /></li><li> eventkey = headers.get(key_hdr);<br /></li><li><br /></li><li> if (logger.isdebugenabled()) {<br /></li><li> logger.debug("{event} " + eventtopic + " : " + eventkey + " : "<br /></li><li> + eventbody);<br /></li><li> }<br /></li><li> <br /></li><li> producerdata<string, message> data = new producerdata<string, message><br /></li><li> (eventtopic, new message(tempbody));<br /></li><li> <br /></li><li> long starttime = system.nanotime();<br /></li><li> logger.debug(eventtopic+"++++"+eventbody);<br /></li><li> producer.send(data);<br /></li><li> long endtime = system.nanotime(); </li><li> }<br /></li><li>} else {<br /></li><li>long processedevents = 0;<br /></li><li>for (; processedevents < batchsize; processedevents += 1) {<br /></li><li>event = ch.take();<br /></li><li><br /></li><li> if (event == null) {<br /></li><li> break;<br /></li><li> }<br /></li><li><br /></li><li> byte[] tempbody = event.getbody();<br /></li><li> string eventbody = new string(tempbody,"utf-8");<br /></li><li> map<string, string> headers = event.getheaders();<br /></li><li><br /></li><li> if ((eventtopic = headers.get(topic_hdr)) == null) {<br /></li><li> eventtopic = topic;<br /></li><li> }<br /></li><li><br /></li><li> eventkey = headers.get(key_hdr);<br /></li><li><br /></li><li> if (logger.isdebugenabled()) {<br /></li><li> logger.debug("{event} " + eventtopic + " : " + eventkey + " : "<br /></li><li> + eventbody);<br /></li><li> logger.debug("event #{}", processedevents);<br /></li><li> }<br /></li><li><br /></li><li> // create a message and add to buffer<br /></li><li> producerdata<string, string> data = new producerdata<string, string><br /></li><li> (eventtopic, eventbody);<br /></li><li> messagelist.add(data);<br /></li><li>}<br /></li><li><br /></li><li>// publish batch and commit.<br /></li><li> if (processedevents > 0) {<br /></li><li> long starttime = system.nanotime(); </li><li> long endtime = system.nanotime(); </li><li> }<br /></li><li>}<br /></li><li><br /></li><li>transaction.commit();<br /></li><li>} catch (exception ex) {<br /></li><li>string errormsg = "failed to publish events";<br /></li><li>logger.error("failed to publish events", ex);<br /></li><li>status = status.backoff;<br /></li><li>if (transaction != null) {<br /></li><li>try {<br /></li><li>transaction.rollback(); </li><li>} catch (exception e) {<br /></li><li>logger.error("transaction rollback failed", e);<br /></li><li>throw throwables.propagate(e);<br /></li><li>}<br /></li><li>}<br /></li><li>throw new eventdeliveryexception(errormsg, ex);<br /></li><li>} finally {<br /></li><li>if (transaction != null) {<br /></li><li>transaction.close();<br /></li><li>}<br /></li><li>}<br /></li><li><br /></li><li>return status; </li></ol> 下一步，修改flume配置文件，将其中sink部分的配置改成kafka sink，如：<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>producer.sinks.r.type = org.apache.flume.sink.kafka.kafkasink<br /> </li><li>producer.sinks.r.brokerlist = bigdata-node00:9092<br /></li><li>producer.sinks.r.requiredacks = 1<br /></li><li>producer.sinks.r.batchsize = 100<br /></li><li>#producer.sinks.r.kafka.producer.type=async<br /></li><li>#producer.sinks.r.kafka.customer.encoding=utf-8<br /></li><li>producer.sinks.r.topic = testflume1</li></ol>type指向kafkasink所在的完整路径
下面的参数都是kafka的一系列参数，最重要的是brokerlist和topic参数
现在重新启动flume，就可以在kafka的对应topic下查看到对应的日志
