In one of the recent use case, we had to implement a complex event processing in real time mode. Storm is used as real time processing engine, but since It doesn’t provide batching of events therefore we took upon Esper to do the required job.<\/p>\n
Esper can be thought as a complex event processing (CEP) component generally used for event series analysis. Complex event processing (CEP) delivers high-speed processing of many events across all the layers identifying the most meaningful events within the event cloud, analyzing their impact, and taking subsequent action in real time.<\/p>\n
In this use case, we get a real time stream from Kafka, which passes through Esper for event processing.<\/p>\n
Steps to configure esper:<\/p>\n
Here is the source code of bean class and bolt.<\/p>\n
EsperTestBean<\/p>\n
[java]
\npublic class EsperTestBean {<\/p>\n
String line;<\/p>\n
public String getLine() {
\n return line;
\n }<\/p>\n
public void setLine(String line) {
\n this.line = line;
\n }<\/p>\n
public static EsperTestBean parse(String line) {
\n EsperTestBean esperTestBean = new EsperTestBean();
\n esperTestBean.setLine(line);
\n return esperTestBean;
\n }
\n}
\n[\/java]<\/p>\n
EsperBolt:<\/p>\n
[java]
\nimport backtype.storm.task.OutputCollector;
\nimport backtype.storm.task.TopologyContext;
\nimport backtype.storm.topology.OutputFieldsDeclarer;
\nimport backtype.storm.topology.base.BaseRichBolt;
\nimport backtype.storm.tuple.Tuple;
\nimport com.espertech.esper.client.*;
\nimport org.apache.commons.logging.Log;
\nimport org.apache.commons.logging.LogFactory;<\/p>\n
import java.util.List;
\nimport java.util.Map;<\/p>\n
public class ProcessLineEsperBolt extends BaseRichBolt {<\/p>\n
private static final long serialVersionUID = 1L;
\n private static final Log log = LogFactory.getLog(ProcessLineEsperBolt.class);<\/p>\n
private OutputCollector collector;
\n private EPServiceProvider epService;<\/p>\n
@SuppressWarnings("rawtypes")
\n @Override
\n public void prepare(Map stormConf, TopologyContext context,
\n OutputCollector collector) {
\n this.collector = collector;
\n this.setUpEsper();
\n }<\/p>\n
private void setUpEsper() {<\/p>\n
\/\/Register the BeanClass as an EventType
\n Configuration configuration = new Configuration();
\n configuration.addEventType("EsperTestBean", EsperTestBean.class.getName());<\/p>\n
epService = EPServiceProviderManager.getDefaultProvider(configuration);
\n epService.initialize();<\/p>\n
\/\/Processing events in batch of every 1 minute using bean class fields
\n EPStatement visitorsStatement = epService.getEPAdministrator().
\n createEPL("select line as found from EsperTestBean.win:time(1 min) output snapshot every 1 minute");
\n visitorsStatement.addListener(new UpdateListener() {<\/p>\n
\/\/ Listener that provides the new events and old events array.
\n @Override
\n public void update(EventBean[] newEvents, EventBean[] oldEvents) {
\n if (newEvents != null) {
\n System.out.println("Batch Length::::::::::::::::::::::" + newEvents.length);
\n for (EventBean e : newEvents) {
\n System.out.println("online ———————————————-line: " + e.get("found"));
\n }
\n }
\n }<\/p>\n
});
\n }<\/p>\n
@Override
\n public void execute(Tuple input) {
\n List<Object> values = input.getValues();
\n epService.getEPRuntime().sendEvent(values.get(0));
\n collector.ack(input);
\n }<\/p>\n
@Override
\n public void declareOutputFields(OutputFieldsDeclarer declarer) {<\/p>\n
}<\/p>\n
}
\n[\/java]<\/p>\n
In above mentioned code, we process stream for every 1 minute. Esper provide the different type of event processing method like window, time based and many more.<\/p>\n
Hope this blog will help you to implement the event processing using Esper.<\/p>\n