In a previous post I mentioned my recent enthusiasm for Complex Event Processing (CEP). In short, Complex Event Processors take as an input a stream of data and either redirect parts of it to listeners according to a set of predefined rules, or trigger events whenever a pattern is found in the data. This is particularly useful in situations where massive amounts of data generated need to be analyzed in real-time.
A really brilliant piece of software that allows you to do just that is called ESPER. You can find the website of the project right here. Esper provides programmers with a language called EPL, mildly similar to SQL, that facilitate the configuration of the rules and patterns that will be applied on the data stream.
The documentation that comes with ESPER is quite complete, but it lacks practical examples, which makes it look like Esper is complicated to use. So here it is, a five minutes guide to Esper. Esper is available in Java, and C#, although the example that follows is written in Java. I assume that you have already downloaded Esper, which can be done by clicking on this link. After unzipping the file you have just downloaded, you should have a folder called esper-3.1.0 somewhere on your disk.
Before getting started, you need to add a couple of libraries to your project. Of course, esper-3.1.0.jar is one of them, but you also need four third-party libraries that are all located in the esper-3.1.0/esper/lib folder.
The five minutes start now. The data we would like to analyze need to be structured in the form of objects before it can be thrown down the pipe to our CEP engine. Let us use a simple example and write a class (some may call it a bean) that represents a stock price at a given time (a stock tick).
import java.util.Date;
public static class Tick {
String symbol;
Double price;
Date timeStamp;
public Tick(String s, double p, long t) {
symbol = s;
price = p;
timeStamp = new Date(t);
}
public double getPrice() {return price;}
public String getSymbol() {return symbol;}
public Date getTimeStamp() {return timeStamp;}
@Override
public String toString() {
return "Price: " + price.toString() + " time: " + timeStamp.toString();
}
}
It has three properties: a symbol name (say “AAPL” for Apple’s stock), a price and a time stamp. Before we start generating billions of ticks, we need to inform the engine about the the kind of objects it will have to handle. This is done via the “Configuration” object that is used when instantiating the CEP engine.
import com.espertech.esper.client.*;
public class main {
public static void main(String [] args){
//The Configuration is meant only as an initialization-time object.
Configuration cepConfig = new Configuration();
// We register Ticks as objects the engine will have to handle
cepConfig.addEventType("StockTick",Tick.class.getName());
// We setup the engine
EPServiceProvider cep = EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);
}
}
For testing purposes, we now create a function that generates random ticks, and throw them down the pipe to the CEP engine. We call this function “GenerateRandomTick”. It takes an EPRuntime object as an argument. This object is used to transmit events to the CEP engine.
import java.util.Random;
import com.espertech.esper.client.*;
public class exampleMain {
private static Random generator=new Random();
public static void GenerateRandomTick(EPRuntime cepRT){
double price = (double) generator.nextInt(10);
long timeStamp = System.currentTimeMillis();
String symbol = "AAPL";
Tick tick= new Tick(symbol,price,timeStamp);
System.out.println("Sending tick:" + tick);
cepRT.sendEvent(tick);
}
public static void main(String[] args) {
//The Configuration is meant only as an initialization-time object.
Configuration cepConfig = new Configuration();
cepConfig.addEventType("StockTick",Tick.class.getName());
EPServiceProvider cep=EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);
EPRuntime cepRT = cep.getEPRuntime();
}
}
Now we have a working CEP engine and a fake data feed, it is time to create our first rule, or in the Esper parlance, our first EPL statement. To do so we need to ask the administrator of the engine to record our statement. The CEP engine will then filter the data it receives, and trigger events whenever that data meets the selection rule, or fulfills the pattern defined in the statement.
public static void main(String[] args) {
//The Configuration is meant only as an initialization-time object.
Configuration cepConfig = new Configuration();
cepConfig.addEventType("StockTick",Tick.class.getName());
EPServiceProvider cep = EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);
EPRuntime cepRT = cep.getEPRuntime();
// We register an EPL statement
EPAdministrator cepAdm = cep.getEPAdministrator();
EPStatement cepStatement = cepAdm.createEPL("select * from " +
"StockTick(symbol='AAPL').win:length(2) " +
"having avg(price) > 6.0");
}
Here, an event in the form of a stock tick will be triggered every time the average over the last 2 ticks is above the value of 6.0.
The next step now consists in creating a listener and connect it to the events generated by our selection rule. This can be done as follows:
cepStatement.addListener(new CEPListener());
There are different ways to implement listener, although the following is the simplest one. Here the listener simply prints the object it has received from the engine:
public static class CEPListener implements UpdateListener {
public void update(EventBean[] newData, EventBean[] oldData) {
System.out.println("Event received: "
+ newData[0].getUnderlying());
}
}
So far, so good. It is now time to test our code. let’s generate a few ticks and see if things work. This can be done by adding the following line at the end of the main function:
for(int i = 0; i< 5; i++) GenerateRandomTick(cepRT); [/code] The code now looks like this (I have placed the Tick class inside the main program so that you should be able to cut and paste the following in one file and run it): [code language="java"] import com.espertech.esper.client.*; import java.util.Random; import java.util.Date; public class exampleMain { public static class Tick { String symbol; Double price; Date timeStamp; public Tick(String s, double p, long t) { symbol = s; price = p; timeStamp = new Date(t); } public double getPrice() {return price;} public String getSymbol() {return symbol;} public Date getTimeStamp() {return timeStamp;} @Override public String toString() { return "Price: " + price.toString() + " time: " + timeStamp.toString(); } } private static Random generator = new Random(); public static void GenerateRandomTick(EPRuntime cepRT) { double price = (double) generator.nextInt(10); long timeStamp = System.currentTimeMillis(); String symbol = "AAPL"; Tick tick = new Tick(symbol, price, timeStamp); System.out.println("Sending tick:" + tick); cepRT.sendEvent(tick); } public static class CEPListener implements UpdateListener { public void update(EventBean[] newData, EventBean[] oldData) { System.out.println("Event received: " + newData[0].getUnderlying()); } } public static void main(String[] args) { //The Configuration is meant only as an initialization-time object. Configuration cepConfig = new Configuration(); cepConfig.addEventType("StockTick", Tick.class.getName()); EPServiceProvider cep = EPServiceProviderManager.getProvider("myCEPEngine", cepConfig); EPRuntime cepRT = cep.getEPRuntime(); EPAdministrator cepAdm = cep.getEPAdministrator(); EPStatement cepStatement = cepAdm.createEPL("select * from " + "StockTick(symbol='AAPL').win:length(2) " + "having avg(price) > 6.0″);
cepStatement.addListener(new CEPListener());
// We generate a few ticks…
for (int i = 0; i < 5; i++) {
GenerateRandomTick(cepRT);
}
}
}
[/code]
Output:
[code language="java"]
log4j:WARN No appenders could be found for logger (com.espertech.esper.epl.metric.MetricReportingPath).
log4j:WARN Please initialize the log4j system properly.
Sending tick:Price: 6.0 time: Tue Jul 21 01:11:15 CEST 2009
Sending tick:Price: 0.0 time: Tue Jul 21 01:11:15 CEST 2009
Sending tick:Price: 7.0 time: Tue Jul 21 01:11:15 CEST 2009
Sending tick:Price: 4.0 time: Tue Jul 21 01:11:15 CEST 2009
Sending tick:Price: 9.0 time: Tue Jul 21 01:11:15 CEST 2009
Event received: Price: 9.0 time: Tue Jul 21 01:11:15 CEST 2009
[/code]
As you can see, only the last two ticks have an average above the value of 6.0, and therefore only one event was triggered by the engine. Fantastic!
Oh, you maybe worry about the two first lines. Yeah, there is still a bit of an issue to sort out. In fact, Esper uses a logging package called log4j that causes these warnings. It can be configured using a file called log4j.xml which you will find in the /etc folder of one of the example project in esper-3.1.0/examples. I don't think that having a XML configuration file for everything is a very good idea and therefore I have used to following trick to configure the logger. Add the following imports at the beginning of your file:
[code language="java"]
import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.SimpleLayout;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
[/code]
and this in the main function before the rest of your code:
[code language="java"]
public static void main(String [] args){
SimpleLayout layout = new SimpleLayout();
ConsoleAppender appender = new ConsoleAppender(new SimpleLayout());
Logger.getRootLogger().addAppender(appender);
Logger.getRootLogger().setLevel((Level) Level.WARN);
(...)
[/code]
The five minutes are now over.
In my next posts, I will explore EPL statements a little bit more in depth, and provide some code to connect two engines together in order to facilitate what is called "event refinement".
I have plotted the time lag between the emission and reception of the event over a period of roughly two seconds. As one can see, the reception of the eventObject is instantaneous most of the time. Sometimes tough, it seems that the the reception is slightly delayed, by at most one millisecond. The usual suspect in this case would be the garbage collector. However the introduction of "memory leaks" in the program, e.g. by allocating large arrays on the heap everytime that a sound event is fired, does not seem to increase the size of the lags. Therefore, I believe that it would be erroneous to jump to a conclusion too quickly and declare that garbage collection is responsible to delaying the processing of events. A good test would be to write the same piece of code C++ and check whether the time lags disappear. I was actually quite surprised by this observation, as I thought that the garbage collection would be more visible.
CoffeeOneSugar's Blog 