使用flume+kafka+storm构建实时日志分析系统_PHP教程

<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>

<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>

<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>