Esper - Java Event Stream Processor

The Esper engine has been developed to address the requirements of applications that analyze and react to events. Some typical examples of applications are:

What these applications have in common is the requirement to process events (or messages) in real-time or near real-time. This is sometimes referred to as complex event processing (CEP) and event stream analysis. Key considerations for these types of applications are throughput, latency and the complexity of the logic required.

The Esper engine was designed to make it easier to build and extend CEP applications.

Relational databases and the standard query language (SQL) are designed for applications in which most data is fairly static and complex queries are less frequent. Also, most databases store all data on disks (except for in-memory databases) and are therefore optimized for disk access.

To retrieve data from a database an application must issue a query. If an application need the data 10 times per second it must fire the query 10 times per second. This does not scale well to hundreds or thousands of queries per second.

Database triggers can be used to fire in response to database update events. However database triggers tend to be slow and often cannot easily perform complex condition checking and implement logic to react.

In-memory databases may be better suited to CEP applications then traditional relational database as they generally have good query performance. Yet they are not optimized to provide immediate, real-time query results required for CEP and event stream analysis.

The Esper engine works a bit like a database turned upside-down. Instead of storing the data and running queries against stored data, the Esper engine allows applications to store queries and run the data through. Response from the Esper engine is real-time when conditions occur that match queries. The execution model is thus continuous rather then only when a query is submitted.

Esper provides two principal methods or mechanisms to process events: event patterns and event stream queries.

Esper offers an event pattern language to specify expression-based event pattern matching. Underlying the pattern matching engine is a state machine implementation. This method of event processing matches expected sequences of presence or absence of events or combinations of events. It includes time-based correlation of events.

Esper also offers event stream queries that address the event stream analysis requirements of CEP applications. Event stream queries provide the windows, aggregation, joining and analysis functions for use with streams of events. These queries are following the EQL syntax. EQL has been designed for similarity with the SQL query language but differs from SQL in its use of views rather then tables. Views represent the different operations needed to structure data in an event stream and to derive data from an event stream.

Esper provides these two methods as alternatives through the same API.

An instance of net.esper.client.Configuration represents all configuration parameters. The Configuration is used to build an (immutable) EPServiceProvider, which provides the administrative and runtime interfaces for an Esper engine instance.

You may obtain a Configuration instance by instantiating it directly and adding or setting values on it. The Configuration instance is then passed to EPServiceProviderManager to obtain a configured Esper engine.

Configuration configuration = new Configuration();
configuration.addEventTypeAlias("PriceLimit", PriceLimit.class.getName());
configuration.addEventTypeAlias("StockTick", StockTick.class.getName());
configuration.addImport("org.mycompany.mypackage.MyUtility");
configuration.addImport("org.mycompany.util.*");

EPServiceProvider epService = EPServiceProviderManager.getProvider("sample", configuration);

Note that Configuration is meant only as an initialization-time object. The Esper engine represented by an EPServiceProvider is immutable and does not retain any association back to the Configuration.

An alternative approach to configuration is to specify a configuration in an XML file.

The default name for the XML configuration file is esper.cfg.xml. Esper reads this file from the root of the CLASSPATH as an application resource via the configure method.

Configuration configuration = new Configuration();		
configuration.configure();

The Configuration class can read the XML configuration file from other sources as well. The configure method accepts URL, File and String filename parameters.

Configuration configuration = new Configuration();		
configuration.configure("myengine.esper.cfg.xml");

Here is an example configuration file. The schema for the configuration file can be found in the etc folder and is named esper-configuration-1-0.

<?xml version="1.0" encoding="UTF-8"?>
<esper-configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
           xsi:noNamespaceSchemaLocation="esper-configuration-1-0.xsd">
  <event-type alias="StockTick" class="net.esper.example.stockticker.event.StockTick"/>
  <event-type alias="PriceLimit" class="net.esper.example.stockticker.event.PriceLimit"/>
  <auto-import import-name="org.mycompany.mypackage.MyUtility"/>
  <auto-import import-name="org.mycompany.util.*"/>
</esper-configuration>		

The example above is only a subset of the configuration items available. The next chapters outline the available configuration in greater detail.

Esper has 2 primary interfaces that this section outlines: The administrative interface and the runtime interface.

Use Esper's administrative interface to create event patterns and EQL statements as discussed in Section 5.1, “Event Pattern Overview” and Section 6.1, “EQL Introduction”.

Use Esper's runtime interface to send events into the engine, emit events and get statistics for an engine instance.

The JavaDoc documentation is also a great source for API information.

Each instance of an Esper engine is completely independent of other engine instances and has its own administrative and runtime interface.

An instance of the Esper engine is obtained via static methods on the EPServiceProviderManager class. The getDefaultProvider method and the getProvider(String URI) methods return an instance of the Esper engine. The latter can be used to obtain multiple instances of the engine for different URI values. The EPServiceProviderManager determines if the URI matches all prior URI values and returns the same engine instance for the same URI value. If the URI has not been seen before, it creates a new engine instance.

The code snipped below gets the default instance Esper engine. Subsequent calls to get the default engine instance return the same instance.

EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();

This code snippet gets an Esper engine for URI RFIDProcessor1. Subsequent calls to get an engine with the same URI return the same instance.

EPServiceProvider epService = EPServiceProviderManager.getProvider("RFIDProcessor1");

An existing Esper engine instance can be reset via the initialize method on the EPServiceProvider instance. This stops and removes all statements in the Engine.

Create event pattern expression and EQL statements via the administrative interface EPAdministrator.

This code snippet gets an Esper engine then creates an event pattern and an EQL statement.

EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
EPAdministrator admin = epService.getEPAdministrator();

EPStatement 10secRecurTrigger = admin.createPattern(
  "every timer:at(*, *, *, *, *, */10)");

EPStatement countStmt = admin.createEQL(
  "select count(*) from MarketDataBean.win:time(60 sec)");

Note that event pattern expressions can also occur within EQL statements. This is outlined in more detail in Section 6.4.2, “Pattern-based event streams”.

The createPattern and createEQL methods return EPStatement instances. Statements are automatically started and active when created. A statement can also be stopped and started again via the stop and start methods shown in the code snippet below.

countStmt.stop();
countStmt.start();

We can subscribe to updates posted by a statement via the addListener and removeListener methods the EPStatement statement. We need to provide an implementation of the UpdateListener interface to the statement.

UpdateListener myListener = new MyUpdateListener();
countStmt.addListener(myListener);

EQL statements and event patterns publish old data and new data to registered UpdateListener listeners. New data published by statements is the events representing the new values of derived data held by the statement. Old data published by statements constists of the events representing the prior values of derived data held by the statement.

Subscribing to events posted by a statement is following a push model. The engine pushes data to listeners when events are received that cause data to change or patterns to match. Alternatively, statements can also serve up data in a pull model via the iterator method. This can come in handy if we are not interested in all new updates, but only want to perform a frequent poll for the latest data. For example, an event pattern that fires every 5 seconds could be used to pull data from an EQL statement. The code snippet below demonstrates some pull code.

Iterator<EventBean> eventIter = countStmt.iterator();
for (EventBean event : eventIter) {
   // .. do something ..
}

This is a second example:

double averagePrice = (Double) eqlStatement.iterator().next().get("average");

Esper places the following restrictions on the pull API and usage of the iterator method:

The EPRuntime interface is used to send events for processing into an Esper engine, and to emit Events from an engine instance to the outside world.

The below code snippet shows how to send a Java object event to the engine. Note that the sendEvent method is overloaded. As events can take on different representation classes in Java, the sendEvent takes parameters to reflect the different types of events that can be send into the engine. The Chapter 4, Event Representations section explains the types of events accepted.

EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
EPRuntime runtime = epService.getEPRuntime();

// Send an example event containing stock market data
runtime.sendEvent(new MarketDataBean('IBM', 75.0));		

Another important method in the runtime interface is the route method. This method is designed for use by UpdateListener implementations that need to send events into an engine instance.

The emit and addEmittedListener methods can be used to emit events from a runtime to a registered set of one or more emitted event listeners. This mechanism is available as a service to enable channel-based publish-subscribe of events emitted from an engine instance via the emit method. Emitting events is not integrated with EQL and is available only via the EPRuntime interface. Events are emitted on an event channel identified by a name. Listeners are implementations of the EmittedListener interface. Via the addEmittedListener method a listener can be added to the specified event channel. The lister receives only those events posted to that channel. The channel parameter to addEmittedListener also allows null values. If a null channel value is specified, the listeners receives emitted events posted on any channel.

Special events are provided that can be used to control the time-keeping of an engine instance. There are two models for an engine to keep track of time. Internal clocking is when the engine instance relies on the java.util.Timer class for time tick events. External clocking can be used to supply time ticks to the engine. The latter is useful for testing time-based event sequences or for synchronizing the engine with an external time source.

By default, the Esper engine uses internal time ticks. This behavior can be changed by sending a timer control event to the engine as shown below.

EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
EPRuntime runtime = epService.getEPRuntime();
// switch to external clocking
runtime.sendEvent(new TimerControlEvent(TimerControlEvent.ClockType.CLOCK_EXTERNAL));

// send a time tick
long timeInMillis = System.currentTimeMillis();	// Or get the time somewhere else
runtime.sendEvent(new CurrentTimeEvent(timeInMillis));

The Esper engine posts events to registered UpdateListener instances ('push' method for receiving events). For many statements events can also be pulled from statements via the iterator method. Both pull and push supply EventBean instances representing the events generated by the engine or events supplied to the engine. Each EventBean instance represents an event, with each event being either an artificial event, composite event or an event supplied to the engine via its runtime interface.

The getEventType method supplies an event's event type information represented by an EventType instance. The EventType supplies event property names and types as well as information about the underlying object to the event.

The engine may generate artificial events that contain information derived from event streams. A typical example for artificial events is the events posted for a statement to calculate univariate statistics on an event property. The below example shows such a statement and queries the generated events for an average value.

// Derive univariate statistics on price for the last 100 market data events
String stmt = "select * from MarketDataBean(symbol='IBM').win:length(100).stat:uni('price')";
EPStatement priceStatsView = epService.getEPAdministrator().createEQL(stmt);
priceStatsView.addListener(testListener);

// Example listener code
public class MyUpdateListener implements UpdateListener
{
    public void update(EventBean[] newData, EventBean[] oldData)
    {
        // Interrogate events
        System.out.println("new average price=" + newData[0].get("average");
	}
}

Composite events are events that aggregate one or more other events. Composite events are typically created by the engine for statements that join two event streams, and for event patterns in which the causal events are retained and reported in a composite event. The example below shows such an event pattern.

// Look for a pattern where AEvent follows BEvent
String pattern = "a=AEvent -> b=BEvent";
EPStatement stmt = epService.getEPAdministrator().createPattern(pattern);
stmt.addListener(testListener);

// Example listener code
public class MyUpdateListener implements UpdateListener
{
    public void update(EventBean[] newData, EventBean[] oldData)
    {
        System.out.println("a event=" + newData[0].get("a").getUnderlying());
        System.out.println("b event=" + newData[0].get("b").getUnderlying());
	}
}

An event is an immutable record of a past occurrence of an action or state change. An event can have a set of event properties that supply information about the event. An event also has an underlying Java object type.

In Esper, an event can be represented by any of the following underlying Java objects:

Esper expressions can include simple as well as indexed, mapped and nested event properties. The table below outlines the different types of properties and their syntax in an event expression. This syntax allows statements to query deep JavaBean objects graphs, XML structures and Map events.

name

sensorId

name[index]

sensor[0]

name('key')

sensor('light')

name.nestedname

sensor.value

Combinations are also possible. For example, a valid combination could be person.address('home').street[0].

Plain Java object events are object instances that expose event properties through JavaBean-style getter methods. Events classes or interfaces do not have to be fully compliant to the JavaBean specification; however for the Esper engine to obtain event properties, the required JavaBean getter methods must be present.

Esper supports JavaBean-style event classes that extend a superclass or implement one or more interfaces. Also, Esper event pattern and EQL statements can refer to Java interface classes and abstract classes.

Classes that represent events should be made immutable. As events are recordings of a state change or action that occurred in the past, the relevant event properties should not be changeable. However this is not a hard requirement and the Esper engine accepts events that are mutable as well.

Please see Chapter 2, Configuration on options for naming event types represented by Java object event classes.

public class EmployeeEvent {
	public String getFirstName();
	public Address getAddress(String type);
	public Employee getSubordinate(int index);
	public Employee[] getAllSubordinates();
}

every EmployeeEvent(firstName='myName')
every EmployeeEvent(address('home').streetName='Park Avenue')
every EmployeeEvent(subordinate[0].name='anotherName')
every EmployeeEvent(allSubordinates[1].name='thatName')
every EmployeeEvent(subordinate[0].address('home').streetName='Water Street')

select firstName, address('work'), subordinate[0].name, subordinate[1].name
from EmployeeEvent
where address('work').streetName = 'Park Ave'

Events can also be represented by objects that implement the java.util.Map interface. Event properties of Map events are the values in the map accessible through the get method exposed by the java.util.Map interface.

The engine can process java.util.Map events via the sendEvent(Map map, String eventTypeAlias) method on the EPRuntime interface. Entries in the Map represent event properties. Keys must be of type java.util.String for the engine to be able to look up event property names specified by pattern or EQL statements. Values can be of any type. JavaBean-style objects as values in a Map can also be processed by the engine.

In order to use Map events, the event type name and property names and types must be made known to the engine via Configuration. Please see the examples in Section 2.4.2, “Events represented by java.util.Map”.

The code snippet below creates and processes a Map event. The example assumes the CarLocationUpdateEvent event type alias has been configured.

Map event = new HashMap();
event.put("carId", carId);
event.put("direction", direction);
epRuntime.sendEvent(event, "CarLocUpdateEvent");

The CarLocUpdateEvent can now be used in a statement:

select carId from CarLocUpdateEvent.win:time(1 min) where direction = 1

The engine can also query Java objects as values in a Map event via the nested property syntax. Thus Map events can be used to aggregate multiple datastructures into a single event and query the composite information in a convenient way. The example below demonstrates a Map event with a transaction and an account object.

Map event = new HashMap();
event.put("txn", txn);
event.put("account", account);
epRuntime.sendEvent(event, "TxnEvent");

An example statement could look as follows.

select account.id, account.rate * txn.amount from TxnEvent.win:time(60 sec) group by account.id

Events can also be represented as org.w3c.dom.Node instances and send into the engine via the sendEvent method on EPRuntime. Please note that configuration is required for allowing the engine to map the event type alias to Node element names. See Chapter 2, Configuration.

Esper allows configuring XPath expressions as event properties. You can specify arbitrary XPath functions or expressions and provide a property name by which their result values will be available for use in expressions. For XML documents that follow an XML schema, Esper can load and interrogate your schema and validate event property names and types against the schema information.

Nested, mapped and indexed event properties are also supported in expressions against org.w3c.dom.Node events. Thus XML trees can conveniently be interrogated using the existing event property syntax for querying JavaBean objects, JavaBean object graphs or java.util.Map events.

Let's look at how a sample XML document could be queried, given the sample XML below.

<?xml version="1.0" encoding="UTF-8"?>
<Sensor>
	<ID>urn:epc:1:4.16.36<ID>
	<Observation Command="READ_PALLET_TAGS_ONLY">
		<ID>00000001<ID>
		<Tag>
			<ID>urn:epc:1:2.24.400<ID>
		</Tag>
		<Tag>
			<ID>urn:epc:1:2.24.401<ID>
		</Tag>
	</Observation>
</Sensor>

To configure the engine for processing Sensor documents, simply configure a SensorEvent event type alias for the Sensor element name via Configuration. Now the document can be queried as below.

select ID, Observation.ID, Observation.Command, Observation.Tag[0], countTags
from SensorEvent.win:time(30 sec)

The equivalent XPath expressions to each of the properties are listed below.

By specifying an event property such below:

nestedElement.mappedElement('key').indexedElement[1]

The equivalent XPath expression is as follows:

/simpleEvent/nestedElement/mappedElement[@id='key']/indexedElement[position() = 2]

Event patterns match when an event or multiple events occur that match the pattern's definition. Patterns can also be time-based.

Pattern expressions can consist of filter expressions combined with pattern operators. Expressions can contain further nested pattern expressions by including the nested expression(s) in () round brackets.

There are 5 types of operators:

This chapter outines how to filter events based on their properties.

The simplest form of filter is a filter for events of a given type without any conditions on the event property values. This filter matches any event of that type regardless of the event's properties. The example below is such a filter. Note that this event pattern would stop firing as soon as the first RfidEvent is encountered.

com.mypackage.myevents.RfidEvent

To make the event pattern fire for every RfidEvent and not just the first event, use the every keyword.

every com.mypackage.myevents.RfidEvent

The example above specifies the fully-qualified Java class name as the event type. Via configuration, the event pattern above can be simplified by using the alias that has been defined for the event type. Interfaces and abstract classes are also supported as event types.

every RfidEvent

Interfaces and superclasses are also supported as event types. In the below example IRfidReadable is an interface class.

every org.myorg.rfid.IRfidReadable

The filtering criteria to filter for events with certain event property values are placed within parenthesis after the event type name.

mypackage.RfidEvent(category="Perishable")

The supported filter operators are

Ranges come in the following 4 varieties. The use of round () or square [] bracket dictates whether an endpoint is included or excluded.

Filter criteria are listed in a comma-separated format. In the example below we look for RfidEvent events with a grade property between 1 and 2 (endpoints included), a price less then 1, and a category of "Perishable".

mypackage.RfidEvent(category="Perishable", price<1.00, grade in [1:2])

Filter criteria can also refer to events matching prior named events in the same expression. Below pattern is an example in which the pattern matches once for every RfidEvent that is preceded by an RfidEvent with the same item id.

every A=mypackage.RfidEvent -> B=mypackage.RfidEvent(itemId=A.itemId)

The syntax shown above allows filter criteria to reference prior results by specifying the event name and event property. This syntax can be used with all filter operators.

Some limitations of filters are:

EQL statements are used to derive and aggregate information from one or more streams of events, and to join or merge event streams. This section outlines EQL syntax. It also outlines the built-in views, which are the building blocks for deriving and aggregating information from event streams.

EQL is similar to SQL in its use of the select clause and the where clause. Where EQL differs most from SQL is in the use of tables. EQL replaces tables with the concept of event streams.

EQL statements contain definitions of one or more views. Similar to tables in an SQL statement, views define the data available for querying and filtering. Some views represent windows over a stream of events. Other views derive statistics from event properties, group events or handle unique event property values. Views can be staggered onto each other to build a chain of views. The Esper engine makes sure that views are reused among EQL statements for efficiency.

The built-in set of views is:

Esper can be extended by plugging-in custom developed views.

EQL queries are created and stored in the engine, and publish results as events are received by the engine or timer events occur that match the criteria specified in the query. Events can also be pulled from running EQL queries.

The select clause in an EQL query specifies the event properties or events to retrieve. The from clause in an EQL query specifies the event stream definitions and stream names to use. The where clause in an EQL query specifies search conditions that specify which event or event combination to search for. For example, the following statement returns the average price for IBM stock ticks in the last 30 seconds.

select avg(price) from StockTick.win:time(30 sec) where symbol='IBM'

EQL queries follow the below syntax. EQL queries can be simple queries or more complex queries. A simple select contains only a select clause and a single stream definition. Complex EQL queries can be build that feature a more elaborate select list utilizing expressions, may join multiple streams, may contain a where clause with search conditions and so on.

[insert into insert_into_def]
select select_list
from stream_def [as name] [, stream_def [as name]] [,...]
[where search_conditions]
[group by grouping_expression_list]
[having grouping_search_conditions]
[output output_specification]
[order by order_by_expression_list]

time-period : [day-part] [hour-part] [minute-part] [seconds-part] [milliseconds-part]

day-part : number ("days" | "day")
hour-part : number ("hours" | "hour")
minute-part : number ("minutes" | "minute" | "min")
seconds-part : number ("seconds" | "second" | "sec")
milliseconds-part : number ("milliseconds" | "millisecond" | "msec")

10 seconds
10 minutes 30 seconds
20 sec 100 msec
1 day 2 hours 20 minutes 15 seconds 110 milliseconds
0.5 minutes

The select clause is required in all EQL statements. The select clause can be used to select all properties via the wildcard *, or to specify a list of event properties and expressions. The select clause defines the event type (event property names and types) of the resulting events published by the statement, or pulled from the statement.

The select clause also offers optional istream and rstream keywords to control how events are posted to UpdateListener instances listening to the statement.

The syntax for the select clause is summarized below.

select [rstream | istream] * | expression_list ... 

select * from stream_def

select * from StockTick(symbol='IBM').win:time(30 sec).stat:uni('price')

select event_property [, event_property] [, ...] from stream_def

select count, stdev from StockTick(symbol='IBM').win:length(100).stat:uni('volume')

select expression [, expression] [, ...] from stream_def

select volume * price from StockTick.win:time_batch(30 sec)

select [event property | expression] as identifier [, ...]

select volume * price as volPrice from StockTick.win:length(100)

select rstream * from StockTick.win:time(30 sec)

The from clause is required in all EQL statements. It specifies one or more event streams. Each event stream can optionally be given a name by means of the as syntax.

from stream_def [as name] [, stream_def [as name]] [, ...]

The event stream definition stream_def as shown in the syntax above can consists of either a filter-based event stream definition or a pattern-based event stream definition.

For joins and outer joins, specify two or more event streams. Joins between pattern-based and filter-based event streams are also supported.

Esper supports joins against relational databases for access to historical or reference data as explained in Section 6.12, “Joining Relational Data via SQL”.