Esper primer

Recently, feeling the need for more complex (no pun intended) event processing than what a simple enterprise integration patterns implementation such as Spring Integration provides, I started looking at Esper for more insight into CEP and ESP but from an open source perspective.

Here I reproduce an example that I followed from the Esper documentation. Note that this is the exact example from the Esper tutorial with javadoc also taken from the Esper documentation which I take no credit for. However I reproduce it here as a complete and coherent example for my own reference and anyone else’s and also with one or two fixes to mistakes in the Esper documentation.

Create an event using a java pojo class.

package esper.example;

import java.text.MessageFormat;

/**
 * Java classes are a good choice for representing events, however Map-based or
 * XML event representations can also be good choices depending on your
 * architectural requirements.
 * 
 * A sample Java class that represents an order event is shown below. A simple
 * plain-old Java class that provides getter-methods for access to event
 * properties works best:
 */
public class OrderEvent {

    private String itemName;
    private double price;

    public OrderEvent(String itemName, double price) {
        this.itemName = itemName;
        this.price = price;
    }

    public String getItemName() {
        return itemName;
    }

    public double getPrice() {
        return price;
    }

    public String toString() {
        return MessageFormat.format("OrderEvent [itemName={0}, price={1}]", new Object[] {
                itemName, new Double(price) });
    }

}

Create a main method that will invoke the application by creating a statement, assigning a listener to that statement and publishing some events.

package esper.example;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;

public class Main {

    final static Logger logger = LoggerFactory.getLogger(Main.class);

    public static void main(String[] args) {

        /*
         * Esper runs out of the box and no configuration is required. However
         * configuration can help make statements more readable and provides the
         * opportunity to plug-in extensions and to configure relational
         * database access.
         * 
         * One useful configuration item specifies Java package names from which
         * to take event classes.
         * 
         * This snippet of using the configuration API makes the Java package of
         * the OrderEvent class known to an engine instance:
         */
        Configuration config = new Configuration();
        config.addEventTypeAutoName(OrderEvent.class.getPackage().getName());

        /*
         * A statement is a continuous query registered with an Esper engine
         * instance that provides results to listeners as new data arrives, in
         * real-time, or by demand via the iterator (pull) API.
         * 
         * The next code snippet obtains an engine instance and registers a
         * continuous query. The query returns the average price over all
         * OrderEvent events that arrived in the last 30 seconds:
         */
        EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider(config);
        String expression = "select itemName, avg(price) from OrderEvent.win:time(30 sec)";
        EPStatement statement = epService.getEPAdministrator().createEPL(expression);

        /*
         * By attaching the listener to the statement the engine provides the
         * statement's results to the listener:
         */
        MyListener listener = new MyListener();
        statement.addListener(listener);

        /*
         * The runtime API accepts events for processing. As a statement's
         * results change, the engine indicates the new results to listeners
         * right when the events are processed by the engine.
         * 
         * Sending events is straightforward as well:
         */
        OrderEvent event1 = new OrderEvent("a", 1);
        OrderEvent event2 = new OrderEvent("b", 2);
        OrderEvent event3 = new OrderEvent("c", 3);
        OrderEvent event4 = new OrderEvent("d", 4);

        EPRuntime epRuntime = epService.getEPRuntime();
        epRuntime.sendEvent(event1);
        epRuntime.sendEvent(event2);
        epRuntime.sendEvent(event3);
        epRuntime.sendEvent(event4);

    }

}

Create a listener that will receive events in real time.

package esper.example;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.UpdateListener;

/**
 * Listeners are invoked by the engine in response to one or more events that
 * change a statement's result set. Listeners implement the UpdateListener
 * interface and act on EventBean instances as the next code snippet outlines:
 */
public class MyListener implements UpdateListener {

    final static Logger logger = LoggerFactory.getLogger(UpdateListener.class);

    public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        EventBean event = newEvents[0];
        Object average = event.get("avg(price)");
        Object itemName = ((Map) event.getUnderlying()).get("itemName");
        logger.info("upon arrival of {} the average stood at {}", itemName, average);
    }

}

This produces the output below as expected.

22:34:21.296 [main] INFO  c.e.esper.client.UpdateListener - upon arrival of a the average stood at 1.0
22:34:21.300 [main] INFO  c.e.esper.client.UpdateListener - upon arrival of b the average stood at 1.5
22:34:21.301 [main] INFO  c.e.esper.client.UpdateListener - upon arrival of c the average stood at 2.0
22:34:21.301 [main] INFO  c.e.esper.client.UpdateListener - upon arrival of d the average stood at 2.5

As a bonus if you want to load events from a csv file that can be done as follows.

Create a csv file with column headings and call it, let’s say, simulation.csv.

itemName,price
a,1
b,2
c,3
d,4

Load and publish events directly from the csv file. Awesome.

package esper.example;

import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.espertech.esperio.AdapterInputSource;
import com.espertech.esperio.InputAdapter;
import com.espertech.esperio.csv.CSVInputAdapter;
import com.espertech.esperio.csv.CSVInputAdapterSpec;

public class Main {

    final static Logger logger = LoggerFactory.getLogger(Main.class);

    public static void main(String[] args) {

        Configuration config = new Configuration();
        config.addEventTypeAutoName("esper.example");

        /*
         * declare event type map
         */
        Map typeMap = new HashMap();
        typeMap.put("itemName", String.class);
        typeMap.put("price", double.class);
        config.addEventType("OrderEvent", typeMap);

        EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider(config);
        String expression = "select itemName, avg(price) from OrderEvent.win:time(30 sec)";
        EPStatement statement = epService.getEPAdministrator().createEPL(expression);

        MyListener listener = new MyListener();
        statement.addListener(listener);

        /*
         * publish events from csv file
         */
        AdapterInputSource adapterInputSource = new AdapterInputSource("simulation.csv");
        (new CSVInputAdapter(epService, adapterInputSource, "OrderEvent")).start();

    }

}

This gets a lot more sophisticated with looping and throttling of the adapter source.

Esper, on first impressions, seems like a very powerful tool for event processing with its event processing language (EPL) modelled on SQL and also very lightweight in that it is simply a jar that can be deployed within your application without any further infrastructure. Being new to CEP and Esper I’m very excited indeed by the prospects it opens up for me and I definitely intend to continue looking further into CEP and ESP academically using Esper to see what else is possible.

What will also be interesting going forward is the experience of gradually porting the more complex use cases I had for Spring Integration to using Esper – although I do understand that they are not an exact match – one being a CEP/ESP solution and another being an EIP/inversion of concurrency solution.

Tagged as: cep, esp, esper, java