Esper Reference

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

More information on CEP can be found at FAQ.

Esper requires the following 3rd-party libraries at runtime:

Esper requires the following 3rd-party libraries for running the test suite:

In Esper, an event can be represented by any of the following underlying Java objects (NEsper .NET, see Section H.6, “.NET Event Underlying Objects”):

Sometimes an event can have properties that are itself events. Esper uses the term fragment and fragment type for such event pieces. The best example is a pattern that matches two or more events and the output event contains the matching events as fragments. In other words, output events can be a composite event that consists of further events, the fragments.

Fragments have the same metadata available as their enclosing composite events. The metadata for enclosing composite events contains information about which properties are fragments, or have a property value that can be represented as a fragment and therefore as an event itself.

Fragments and type metadata can allow your application to navigate composite events without the need for using the Java reflection API and reducing the coupling to the underlying event representation. The API is further described in Section 16.6, “Event and Event Type”.

More information on event representations can be found in the appendix. The links are:

sendEvent(Object event)

sendEvent(Map map, String mapEventTypeName)

sendEvent(Object[] objectarray, String objectArrayEventTypeName)

sendEvent(Object avroGenericDataDotRecord, String avroEventTypeName)

sendEvent(org.w3c.dom.Node node)

N/A

@EventRepresentation(map)

@EventRepresentation(objectarray)

@EventRepresentation(avro)

N/A

create schema name as class_name

create map schema name as (...)

create objectarray schema name as (...)

create avro schema name as (...)

N/A

create map schema ParkingEvent as (carId string, driverName string)

insert into ParkingEvent select carId, 'jim' as driverName from CarArrivalEvent

Esper does not require a fixed tuple structure and fully supports generic tuples, i.e. there does not need to be a fixed set of attributes or event properties and event properties can be added and queried at runtime.

The facilities for support of generic tuples are:

There is no need to explicitly create an event type for each tuple type. It is not necessary to create classes for tuple types at all. Events can be arbitrary objects.

The engine validates EPL at statement creation time therefore there is an advantage if type information is available: the engine can verify your EPL statement against the known properties and types, preventing you as the EPL designer from making mistakes in EPL design. The engine does not verify dynamic properties, which may return null at runtime. If type information is not available then properties are assumed to return java.lang.Object-typed values.

For example, let's say we need a generic tuple and we have Map events:

create schema GenericTuple()

Create statements that use dynamic properties, as the next EPL shows, which casts the timestamp value to a long-type value and outputs the hour-minute-second string:

select cast(timestamp?, long).format('hh mm ss') from GenericTuple

Send events like this:

Map<String, Object> genericEvent = new HashMap<>();
genericEvent.put("timestamp", new Date().getTime());
genericEvent.put("some_other_property", "hello");
epService.getEPRuntime().sendEvent(genericEvent, "GenericTuple");

Part of the extension and plug-in features of Esper is an event representation API. This set of classes allow an application to create new event types and event instances based on information available elsewhere, statically or dynamically at runtime when EPL statements are created. Please see Section 19.8, “Event Type And Event Object” for details.

Creating a plug-in event representation can be useful when your application has existing Java classes that carry event metadata and event property values and your application does not want to (or cannot) extract or transform such event metadata and event data into one of the built-in event representations (POJO Java objects, Map, Object-array or XML DOM).

Further use of a plug-in event representation is to provide a faster or short-cut access path to event data. For example, access to event data stored in a XML format through the Streaming API for XML (StAX) is known to be very efficient. A plug-in event representation can also provide network lookup and dynamic resolution of event type and dynamic sourcing of event instances.

Currently, EsperIO provides the following additional event representations:

Please see the EsperIO documentation for details on the above.

The chapter on Section 19.8, “Event Type And Event Object” explains how to create your own custom event representation.

The first means to handle updating events is the update istream clause as further described in Section 5.20, “Updating an Insert Stream: the Update IStream Clause”. It is useful when you need to update events as they enter a stream, before events are evaluated by any particular consuming statement to that stream.

The second means to update events is the on-merge and on-update clauses, for use with tables and named windows only, as further described in Section 6.8, “Triggered Upsert using the On-Merge Clause” and Section 6.6, “Updating Data: the On Update clause”. On-merge is similar to the SQL merge clause and provides what is known as an "Upsert" operation: Update existing events or if no existing event(s) are found then insert a new event, all in one atomic operation provided by a single EPL statement. On-update can be used to update individual properties of rows held in a table or named window.

The third means to handle updating events is the revision event types, for use with named windows only, as further described in Section 6.11, “Versioning and Revision Event Type Use with Named Windows”. With revision event types one can declare, via configuration only, multiple different event types and then have the engine present a merged event type that contains a superset of properties of all merged types, and have the engine merge events as they arrive without additional EPL statements.

Note that patterns do not reflect changes to past events. For the temporal nature of patterns, any changes to events that were observed in the past do not reflect upon current pattern state.

To extract information from a coarse-grained event or to perform bulk operations on the rows of the property graph in an event, Esper provides a convenient syntax: When specifying a filter expression in a pattern or in a select clause, it may contain a contained-event selection syntax, as further described in Section 5.19, “Contained-Event Selection”.

For NEsper .NET also see Section H.10, “.NET Event Objects Instantiated and Populated by Insert Into”.

The insert into clause can populate instantiate new instances of Java object events, java.util.Map events and Object[] (object array) events directly from the results of select clause expressions and populate such instances. Simply use the event type name as the stream name in the insert into clause as described in Section 5.10, “Merging Streams and Continuous Insertion: the Insert Into Clause”.

If instead you have an existing instance of a Java object returned by an expression, such as a single-row function or static method invocation for example, you can transpose that expression result object to a stream. This is described further in Section 5.10.7, “Transposing an Expression Result” and Section 10.4, “Select-Clause transpose Function”.

The column names specified in the select and insert into clause must match available writable properties in the event object to be populated (the target event type). The expression result types of any expressions in the select clause must also be compatible with the property types of the target event type.

If populating a POJO-based event type and the class provides a matching constructor, the expression result types of expressions in the select clause must be compatible with the constructor parameters in the order listed by the constructor. The insert into clause column names are not relevant in this case.

Consider the following example statement:

insert into com.mycompany.NewEmployeeEvent 
select fname as firstName, lname as lastName from HRSystemEvent

The above example specifies the fully-qualified class name of NewEmployeeEvent. The engine instantianes NewEmployeeEvent for each result row and populates the firstName and lastName properties of each instance from the result of select clause expressions. The HRSystemEvent in the example is assumed to have lname and fname properties, and either setter-methods and a default constructor, or a matching constructor.

Note how the example uses the as-keyword to assign column names that match the property names of the NewEmployeeEvent target event. If the property names of the source and target events are the same, the as-keyword is not required.

The next example is an alternate form and specifies property names within the insert into clause instead. The example also assumes that NewEmployeeEvent has been defined or imported via configuration since it does not specify the event class package name:

insert into NewEmployeeEvent(firstName, lastName) 
select fname, lname from HRSystemEvent

Finally, this example populates HRSystemEvent events. The example populates the value of a type property where the event has the value 'NEW' and populates a new event object with the value 'HIRED', copying the fname and lname property values to the new event object:

insert into HRSystemEvent 
select fname, lname, 'HIRED' as type from HRSystemEvent(type='NEW')

The matching of the select or insert into-clause column names to target event type's property names is case-sensitive. It is allowed to only populate a subset of all available columns in the target event type. Wildcard (*) is also allowed and copies all fields of the events or multiple events in a join.

For Java object events, your event class must provide setter-methods according to JavaBean conventions or, alternatively, a matching constructor. If the event class provides setter methods the class should also provide a default constructor taking no parameters. If the event class provides a matching constructor there is no need for setter-methods. If your event class does not have a default constructor and setter methods, or a matching constructor, your application may configure a factory method via ConfigurationEventTypeLegacy. If your event class does not have a default constructor and there is no factory method provided, the engine uses in connection with the Oracle JVM the sun.reflect.ReflectionFactory, noting that in this case member variables do not get initialized to assigned defaults.

The engine follows Java standards in terms of widening, performing widening automatically in cases where widening type conversion is allowed without loss of precision, for both boxed and primitive types and including BigInteger and BigDecimal.

When inserting array-typed properties into a Java, Map-type or Object-array underlying event the event definition should declare the target property as an array.

Please note the following limitations:

For NEsper .NET also see Section H.11, “.NET Processing Model Introduction”.

The Esper processing model is continuous: Update listeners and/or subscribers to statements receive updated data as soon as the engine processes events for that statement, according to the statement's choice of event streams, views, filters and output rates.

As outlined in Chapter 16, API Reference the interface for listeners is com.espertech.esper.client.UpdateListener. Implementations must provide a single update method that the engine invokes when results become available:

A second, strongly-typed and native, highly-performant method of result delivery is provided: A subscriber object is a direct binding of query results to a Java object. The object, a POJO, receives statement results via method invocation. The subscriber class need not implement an interface or extend a superclass. Please see Section 16.3.3, “Setting a Subscriber Object”.

The engine provides statement results to update listeners by placing results in com.espertech.esper.client.EventBean instances. A typical listener implementation queries the EventBean instances via getter methods to obtain the statement-generated results.

The get method on the EventBean interface can be used to retrieve result columns by name. The property name supplied to the get method can also be used to query nested, indexed or array properties of object graphs as discussed in more detail in Chapter 2, Event Representations and Section 16.6, “Event and Event Type”

The getUnderlying method on the EventBean interface allows update listeners to obtain the underlying event object. For wildcard selects, the underlying event is the event object that was sent into the engine via the sendEvent method. For joins and select clauses with expressions, the underlying object implements java.util.Map.

Filters to event streams allow filtering events out of a given stream before events enter a data window (if there are data windows defined in your query). The statement below shows a filter that selects Withdrawal events with an amount value of 200 or more.

select * from Withdrawal(amount>=200)#length(5)

With the filter, any Withdrawal events that have an amount of less then 200 do not enter the length window and are therefore not passed to update listeners. Filters are discussed in more detail in Section 5.4.1, “Filter-based Event Streams” and Section 7.4, “Filter Expressions In Patterns”.

Figure 3.3. Output example for a statement with an event stream filter

The next statement applies a where-clause to Withdrawal events. Where-clauses are discussed in more detail in Section 5.5, “Specifying Search Conditions: the Where Clause”.

select * from Withdrawal#length(5) where amount >= 200

The where-clause applies to both new events and old events. As the diagram below shows, arriving events enter the window however only events that pass the where-clause are handed to update listeners. Also, as events leave the data window, only those events that pass the conditions in the where-clause are posted to listeners as old events.

Figure 3.4. Output example for a statement with where-clause

In this section we explain the output model of statements employing a time window view and a time batch view.

3.5.1. Time Window

A time window is a moving window extending to the specified time interval into the past based on the system time. Time windows enable us to limit the number of events considered by a query, as do length windows.

As a practical example, consider the need to determine all accounts where the average withdrawal amount per account for the last 4 seconds of withdrawals is greater then 1000. The statement to solve this problem is shown below.

select account, avg(amount) 
from Withdrawal#time(4 sec) 
group by account
having amount > 1000

The next diagram serves to illustrate the functioning of a time window. For the diagram, we assume a query that simply selects the event itself and does not group or filter events.

select * from Withdrawal#time(4 sec)

The diagram starts at a given time t and displays the contents of the time window at t + 4 and t + 5 seconds and so on.

Figure 3.5. Output example for a statement with a time window

The activity as illustrated by the diagram:

1. At time t + 4 seconds an event W1 arrives and enters the time window. The engine reports the new event to update listeners.

2. At time t + 5 seconds an event W2 arrives and enters the time window. The engine reports the new event to update listeners.

3. At time t + 6.5 seconds an event W3 arrives and enters the time window. The engine reports the new event to update listeners.

4. At time t + 8 seconds event W1 leaves the time window. The engine reports the event as an old event to update listeners.

3.5.2. Time Batch

The time batch view buffers events and releases them every specified time interval in one update. Time windows control the evaluation of events, as does the length batch window.

The next diagram serves to illustrate the functioning of a time batch view. For the diagram, we assume a simple query as below:

select * from Withdrawal#time_batch(4 sec)

The diagram starts at a given time t and displays the contents of the time window at t + 4 and t + 5 seconds and so on.

Figure 3.6. Output example for a statement with a time batch view

The activity as illustrated by the diagram:

1. At time t + 1 seconds an event W1 arrives and enters the batch. No call to inform update listeners occurs.

2. At time t + 3 seconds an event W2 arrives and enters the batch. No call to inform update listeners occurs.

3. At time t + 4 seconds the engine processes the batched events and a starts a new batch. The engine reports events W1 and W2 to update listeners.

4. At time t + 6.5 seconds an event W3 arrives and enters the batch. No call to inform update listeners occurs.

5. At time t + 8 seconds the engine processes the batched events and a starts a new batch. The engine reports the event W3 as new data to update listeners. The engine reports the events W1 and W2 as old data (prior batch) to update listeners.

For more detailed information on batch windows please see Section 14.2, “A Note on Batch Windows”.

Related to batch data windows is output rate limiting. While batch data windows retain events the output clause offered by output rate limiting can control or stabilize the rate at which events are output, see Section 5.7, “Stabilizing and Controlling Output: the Output Clause”.

Let's look at how a time batch window may be used:

select account, amount from Withdrawal#time_batch(1 sec)

The above statement collects events arriving during a one-second interval, at the end of which the engine posts the collected events as new events (insert stream) to each listener. The engine posts the events collected during the prior batch as old events (remove stream). The engine starts posting events to listeners one second after it receives the first event and thereon.

For statements containing aggregation functions and/or a group by clause, the engine posts consolidated aggregation results for an event batch. For example, consider the following statement:

select sum(amount) as mysum from Withdrawal#time_batch(1 sec)

Note that output rate limiting also generates batches of events following the output model as discussed here.

An event sent by your application or generated by statements is visible to all other statements in the same engine instance. Similarly, current time (the time horizon) moves forward for all statements in the same engine instance. Please see the Chapter 16, API Reference chapter for how to send events and how time moves forward through system time or via simulated time, and the possible threading models.

Within an Esper engine instance you can additionally control event visibility and current time on a statement level, under the term isolated service as described in Section 16.9, “Service Isolation”.

An isolated service provides a dedicated execution environment for one or more statements. Events sent to an isolated service are visible only within that isolated service. In the isolated service you can move time forward at the pace and resolution desired without impacting other statements that reside in the engine runtime or other isolated services. You can move statements between the engine and an isolated service.

This section discusses the notion of context and its role in the Esper event processing language (EPL).

When you look up the word context in a dictionary, you may find: Context is the set of circumstances or facts that surround a particular event, situation, etc..

Context-dependent event processing occurs frequently: For example, consider a requirement that monitors banking transactions. For different customers your analysis considers customer-specific aggregations, patterns or data windows. In this example the context of detection is the customer. For a given customer you may want to analyze the banking transactions of that customer by using aggregations, data windows, patterns including other EPL constructs.

In a second example, consider traffic monitoring to detect speed violations. Assume the speed limit must be enforced only between 9 am and 5 pm. The context of detection is of temporal nature.

A context takes a cloud of events and classifies them into one or more sets. These sets are called context partitions. An event processing operation that is associated with a context operates on each of these context partitions independently. (Credit: Taken from the book "Event Processing in Action" by Opher Etzion and Peter Niblett.)

A context is a declaration of dimension and may thus result in one or more context partitions. In the banking transaction example there the context dimension is the customer and a context partition exists per customer. In the traffic monitoring example there is a single context partition that exists only between 9 am and 5 pm and does not exist outside of that daily time period.

In an event processing glossary you may find the term event processing agent. An EPL statement is an event processing agent. An alternative term for context partition is event processing agent instance.

Esper EPL allows you to declare contexts explicitly, offering the following benefits:

Esper EPL allows you to declare a context explicitly via the create context syntax introduced below.

After you have declared a context, one or more EPL statements can refer to that context by specifying context name. When an EPL statement refers to a context, all EPL-statement related state such as aggregations, patterns or data windows etc. exists once per context partition.

If an EPL statement does not declare a context, it implicitly has a single context partition. The single context partition lives as long as the EPL statement is started and ends when the EPL statement is stopped.

You may have heard of the term session. A context partition is the same as a session.

You may have heard of the term session window to describe the duration between when a session becomes alive to when a session gets destroyed. We use the term context partition lifecycle instead.

The context declaration specifies how the engine manages context partitions (or sessions):

For more information on locking and threading please see Section 16.7, “Engine Threading and Concurrency”. For performance related information please refer to Chapter 22, Performance.

The create context statement declares a context by specifying a context name and context dimension information.

A context declaration by itself does not consume any resources or perform any logic until your application starts at least one statement that refers to that context. Until then the context is inactive and not in use.

When your application creates or starts the first statement that refers to the context, the engine activates the context.

As soon as your application stops or destroys all statements that refer to the context, the context becomes inactive again.

When your application stops or destroys a statement that refers to a context, the context partitions associated to that statement also end (context partitions associated to other started statements live on).

When your application stops or destroys the statement that declared the context and does not also stop or destroy any statements that refer to the context, the context partitions associated to each such statement do not end.

When your application destroys the statement that declared the context and destroys all statements that refer to that context then the engine removes the context declaration entirely.

The create context statement posts no output events to listeners or subscribers and does not return any rows when iterated.

create context context_name partition [by]
  event_property [and event_property [and ...]] from stream_def
  [, event_property [...] from stream_def]
  [, ...]

create context SegmentedByCustomer partition by custId from BankTxn

context SegmentedByCustomer
select custId, account, sum(amount) from BankTxn group by account

context SegmentedByCustomer
select * from pattern [
  every a=BankTxn(amount > 400) -> b=BankTxn(amount > 400) where timer:within(10 minutes)
]

// Neither LoginEvent nor LogoutEvent are listed in the context declaration
context SegmentedByCustomer
select * from pattern [every a=LoginEvent -> b=LogoutEvent where timer:within(10 minutes)]

// Not valid
create context SegmentedByCustomer partition by custId from BankTxn, account from BankTxn

// Invalid: Type mismatch between properties
create context SegmentedByCustomer partition by custId from BankTxn, loginTime from LoginEvent

create context SegmentedByCustomer partition by 
  custId from BankTxn, loginId from LoginEvent, loginId from LogoutEvent

create context SegmentedByCustomer partition by 
  custId from BankTxn, loginId from LoginEvent(failed=false)

create context ByCustomerAndAccount partition by custId and account from BankTxn

context ByCustomerAndAccount select custId, account, sum(amount) from BankTxn

create context ByCustomerAndAccount partition by custId and account from BankTxn

context ByCustomerAndAccount
  select context.name, context.id, context.key1, context.key2 from BankTxn

create context ByCust partition by custId from BankTxn

context ByCust 
  select * from BankTxn as t1 unidirectional, BankTxn#time(30) t2 
  where t1.amount = t2.amount

context ByCust
  select * from SecurityEvent as t1 unidirectional, BankTxn#time(30) t2 
  where t1.customerName = t2.customerName

create context context_name coalesce [by] 
  hash_func_name(hash_func_param) from stream_def
  [, hash_func_name(hash_func_param) from stream_def ]
  [, ...]
  granularity granularity_value
  [preallocate]

create context SegmentedByCustomerHash
  coalesce by consistent_hash_crc32(custId) from BankTxn granularity 16 preallocate

context SegmentedByCustomerHash
select custId, account, sum(amount) from BankTxn group by custId, account

create context MyHashContext
  coalesce by computeHash(*) from BankTxn granularity 16 preallocate

// Neither LoginEvent nor LogoutEvent are listed in the context declaration
context SegmentedByCustomerHash
select * from pattern [every a=LoginEvent -> b=LogoutEvent where timer:within(10 minutes)]

create context HashedByCustomer as coalesce
  consistent_hash_crc32(custId) from BankTxn,
  consistent_hash_crc32(loginId) from LoginEvent, 
  consistent_hash_crc32(loginId) from LogoutEvent
  granularity 32 preallocate

create context HashedByCustomer 
  coalesce consistent_hash_crc32(loginId) from LoginEvent(failed = false) 
  granularity 1024 preallocate

create context ByCustomerHash coalesce consistent_hash_crc32(custId) from BankTxn granularity 1024

context ByCustomerHash
  select context.name, context.id from BankTxn

create context context_name
  group [by] group_expression as category_label
  [, group [by] group_expression as category_label]
  [, ...]
  from stream_def

create context CategoryByTemp
  group temp < 65 as cold,
  group temp between 65 and 85 as normal,
  group temp > 85 as large
  from SensorEvent

context CategoryByTemp select context.label, count(*) from SensorEvent

context CategoryByTemp
  select context.name, context.id, context.label from SensorEvent

create context context_name
  start (@now | start_condition) 
  [ end end_condition ]

create context NineToFive start (0, 9, *, *, *) end (0, 17, *, *, *)

context NineToFive select * from TrafficEvent(speed >= 100)

create context PowerOutage start PowerOutageEvent end pattern [PowerOnEvent -> timer:interval(5)]

context PowerOutage select * from TemperatureEvent

create context Every15Minutes start @now end after 15 minutes

create context AlwaysOn start @now

create context MyCtx start MyStartEvent end MyEndEvent

context MyCtx select count(*) as cnt from MyEndEvent output when terminated

create context context_name
  initiated [by] [distinct (distinct_value_expr [,...])] [@now and] initiating_condition
  [ terminated [by] terminating_condition ]

create context CtxTrainEnter
  initiated by TrainEnterEvent as te
  terminated after 5 minutes

context CtxTrainEnter
select t1 from pattern [
  t1=TrainEnterEvent -> timer:interval(5 min) and not TrainLeaveEvent(trainId = context.te.trainId)
  ]

create context CtxEachMinute
  initiated @now and pattern [every timer:interval(1 minute)]
  terminated after 1 minutes

context CtxEachMinute select avg(temp) from SensorEvent

create context CtxTrainEnter initiated by TrainEnterEvent as te

create context OrderContext
  initiated by distinct(orderId) NewOrderEvent as newOrder 
  terminated by CloseOrderEvent(closeOrderId = newOrder.orderId)

context NineToFive 
select context.name, context.startTime, context.endTime from TrafficEvent(speed >= 100)

context CtxTrainEnter
select *, context.te.trainId, context.id, context.name
from TrainLeaveEvent(trainId = context.te.trainId)

event_stream_name [(filter_criteria)] [as stream_name]

// A non-overlapping context that starts when MyStartEvent arrives and ends when MyEndEvent arrives
create context MyContext start MyStartEvent end MyEndEvent

// An overlapping context where each MyEvent with level greater zero 
// initiates a new context partition that terminates after 10 seconds
create context MyContext initiated MyEvent(level > 0) terminated after 10 seconds

// A non-overlapping context that starts when MyEvent arrives
// and ends when a matching MyEvent arrives (same id)
create context MyContext 
start MyEvent as myevent
end MyEvent(id=myevent.id)

// An overlapping context where each MyInitEvent initiates a new context partition 
// that terminates when a matching MyTermEvent arrives 
create context MyContext 
initiated by MyInitEvent as e1 
terminated by MyTermEvent(id=e1.id, level <> e1.level)

pattern [pattern_expression] [@inclusive]

// A non-overlapping context that starts when either StartEventOne or StartEventTwo arrive
// and that ends after 5 seconds.
// Here neither StartEventOne or StartEventTwo count towards any statements
// that are referring to the context.
create context MyContext 
  start pattern [StartEventOne or StartEventTwo] 
  end after 5 seconds

// Same as above. 
// Here both StartEventOne or StartEventTwo do count towards any statements
// that are referring to the context.
create context MyContext 
  start pattern [a=StartEventOne or b=StartEventTwo] @inclusive 
  end after 5 seconds

// An overlapping context where each distinct MyInitEvent initiates a new context 
// and each context partition terminates after 20 seconds
// We use @inclusive to say that the same MyInitEvent that fires the pattern
// also applies to statements that are associated to the context.
create context MyContext
  initiated by pattern [every-distinct(a.id, 20 sec) a=MyInitEvent]@inclusive
  terminated after 20 sec

// An overlapping context where each pattern match initiates a new context 
// and all context partitions terminate when MyTermEvent arrives.
// The MyInitEvent and MyOtherEvent that trigger the pattern are themselves not included 
// in any statements that are associated to the context.
create context MyContext
  initiated by pattern [every MyInitEvent -> MyOtherEvent where timer:within(5)]
  terminated by MyTermEvent

// A non-overlapping context that starts when either StartEventOne or StartEventTwo arrive
// and that ends when either a matching EndEventOne or EndEventTwo arrive
create context MyContext 
  start pattern [a=StartEventOne or b=StartEventTwo]@inclusive
  end pattern [EndEventOne(id=a.id) or EndEventTwo(id=b.id)]

// A non-overlapping context started daily between 9 am to 5 pm
// and not started outside of these hours:
create context NineToFive start (0, 9, *, *, *) end (0, 17, *, *, *)

// An overlapping context where crontab initiates a new context every 1 minute
// and each context partition terminates after 10 seconds:
create context MyContext initiated (*, *, *, *, *) terminated after 10 seconds

after time_period_expression

// A non-overlapping context started after 10 seconds 
// that ends 1 minute after it starts and that again starts 10 seconds thereafter.
create context NonOverlap10SecFor1Min start after 10 seconds end after 1 minute

// An overlapping context that starts a new context partition every 5 seconds
// and each context partition lasts 1 minute
create context Overlap5SecFor1Min initiated after 5 seconds terminated after 1 minute

A nested context is a context that is composed from two or more contexts.

The syntax for creating a nested context is as follows:

create context context_name
  context nested_context_name [as] nested_context_definition ,
  context nested_context_name [as] nested_context_definition [, ...]

The context_name you assign to the context can be any identifier.

Following the context name is a comma-separated list of nested contexts. For each nested context specify the context keyword followed a nested context name and the nested context declaration. Any of the context declarations as outlined in Section 4.2, “Context Declaration” are allowed for nested contexts. The order of nested context declarations matters as outlined below. The nested context names have meaning only in respect to built-in properties and statements may not be assigned to nested context names.

The next statement creates a nested context NineToFiveSegmented that, between 9 am and 5 pm, allocates a new context partition for each customer id:

create context NineToFiveSegmented
  context NineToFive start (0, 9, *, *, *) end (0, 17, *, *, *),
  context SegmentedByCustomer partition by custId from BankTxn

The following statement refers to the nested context to compute a total withdrawal amount per account for each customer but only between 9 am and 5 pm:

context NineToFiveSegmented
select custId, account, sum(amount) from BankTxn group by account

Esper implements nested contexts as a context tree: The context declared first controls the lifecycle of the context(s) declared thereafter. Thereby, in the above example, outside of the 9am-to-5pm time the engine has no memory and consumes no resources in relationship to bank transactions or customer ids.

When combining segmented contexts, the set of context partitions for the nested context effectively is the Cartesian product of the partition sets of the nested segmented contexts.

When combining temporal contexts with other contexts, since temporal contexts may overlap and may terminate, it is important to understand that temporal contexts control the lifecycle of sub-contexts (contexts declared thereafter). The order of declaration of contexts in a nested context can thereby change resource usage and output result.

The next statement creates a context that allocates context partition only when a train enters a station and then for each hash of the tag id of a passenger as indicated by PassengerScanEvent events, and terminates all context partitions after 5 minutes:

create context CtxNestedTrainEnter
  context InitCtx initiated by TrainEnterEvent as te terminated after 5 minutes,
  context HashCtx coalesce by consistent_hash_crc32(tagId) from PassengerScanEvent
    granularity 16 preallocate

In the example above the engine does not start tracking PassengerScanEvent events or hash codes or allocate context partitions until a TrainEnterEvent arrives.

create context CtxSampleNestedContext
  context SpanA start AStart end AEnd,
  context SpanB start BStart end BEnd

context CtxSampleNestedContext select count(*) from C

create context CtxSampleNestedContext 
  start pattern[every a=AStart or every a=BStart] as mypattern 
  end pattern[every AEnd or every BEnd]

context CtxNestedTrainEnter
select context.InitCtx.te.trainId, context.HashCtx.id,
  tagId, count(*) from PassengerScanEvent group by tagId

context NineToFiveSegmented
select context.NineToFive.startTime, context.SegmentedByCustomer.key1 from BankTxn

context NineToFiveSegmented select context.name, context.id from BankTxn

You do not need to declare a context to partition data windows, aggregation values or patterns themselves individually. You may mix-and-match partitioning as needed.

The table below outlines other partitioning syntax supported by EPL:

// Length window of 2 events per customer
select * from
  BankTxn#groupwin(custId)#length(2)

select avg(price), window(*)
  from BankTxn group by custId

select * from pattern [
  every a=BankTxn -> BankTxn(custId = a.custId)...]

select * from match_recognize 
  ... partition by custId

select * from ... where a.custId = b.custId

You may use output rate limiting to trigger output when a context partition ends, as further described in Section 5.7, “Stabilizing and Controlling Output: the Output Clause”.

Consider the fixed temporal context: A new context partition gets allocated at the designated start time and the current context partition ends at the designated end time. To trigger output when the context partition ends and before it gets removed, read on.

The same is true for the initiated temporal context: That context starts a new context partition when trigger events arrive or when a pattern matches. Each context partition expires (ends, terminates) after the specified time period passed. To trigger output at the time the context partition expires, read on.

You may use the when terminated syntax with output rate limiting to trigger output when a context partition ends. The following example demonstrates the idea by declaring an initiated temporal context.

The next statement creates a context CtxEachMinute that initiates a new context partition every 1 minute, and that expires each context partition after 5 minutes:

create context CtxEachMinute
initiated by pattern [every timer:interval(1 min)]
terminated after 5 minutes

The following statement computes an ongoing average temperature however only outputs the last value of the average temperature after 5 minutes when a context partition ends:

context CtxEachMinute
select context.id, avg(temp) from SensorEvent output snapshot when terminated

The when terminated syntax can be combined with other output rates.

The next example outputs every 1 minute and also when the context partition ends:

context CtxEachMinute
select context.id, avg(temp) from SensorEvent output snapshot every 1 minute and when terminated

In the case that the end/terminating condition of the context partition is an event or pattern, the context properties contain the information of the tagged events in the pattern or the single event that ended/terminated the context partition.

For example, consider the following context wherein the engine initializes a new context partition for each arriving MyStartEvent event and that terminates a context partition when a matching MyEndEvent arrives:

create context CtxSample
initiated by MyStartEvent as startevent
terminated by MyEndEvent(id = startevent.id) as endevent

The following statement outputs the id property of the initiating and terminating event and only outputs when a context partition ends:

context CtxSample 
select context.startevent.id, context.endevent.id, count(*) from MyEvent
output snapshot when terminated

You may in addition specify a termination expression that the engine evaluates when a context partition terminates. Only when the terminaton expression evaluates to true does output occur. The expression may refer to built-in properties as described in Section 5.7.1.1, “Controlling Output Using an Expression”. The syntax is as follows:

...output when terminated and termination_expression

The next example statement outputs when a context partition ends but only if at least two events are available for output:

context CtxEachMinute
select * from SensorEvent output when terminated and count_insert >= 2

The final example EPL outputs when a context partition ends and sets the variable myvar to a new value:

context CtxEachMinute
select * from SensorEvent output when terminated then set myvar=3

Named windows are globally-visible data windows that may be referred to by multiple statements. You may refer to named windows in statements that declare a context without any special considerations, with the exception of on-action statements (latter must refer to the same context associated with the named window).

You may also create a named window and declare a context for the named window. In this case the engine in effect manages separate named windows, one for each context partition. Limitations apply in this case that we discuss herein.

For example, consider the 9 am to 5 pm fixed temoral context as shown earlier:

create context NineToFive start (0, 9, *, *, *) end (0, 17, *, *, *)

You may create a named window that only exists between 9 am and 5 pm:

context NineToFive create window SpeedingEvents1Hour#time(30 min) as TrafficEvent

You can insert into the named window:

insert into SpeedingEvents1Hour select * from TrafficEvent(speed > 100)

Any on-merge, on-select, on-update and on-delete statements must however declare the same context.

The following is not a valid statement as it does not declare the same context that was used to declare the named window:

// You must declare the same context for on-trigger statements
on TruncateEvent delete from SpeedingEvents1Hour

The following is valid:

context NineToFive on TruncateEvent delete from SpeedingEvents1Hour

For context declarations that require specifying event types, such as the hash segmented context and keyed segmented context, please provide the named window underlying event type.

The following sample EPL statements define a type for the named window, declare a context and associate the named window to the context:

create schema ScoreCycle (userId string, keyword string, productId string, score long)

create context HashByUserCtx as 
  coalesce by consistent_hash_crc32(userId) from ScoreCycle granularity 64

context HashByUserCtx create window ScoreCycleWindow#unique(productId, keyword) as ScoreCycle

Tables are globally-visible data structures that hold rows organized by primary key(s) and that may be referred to by multiple statements. You may refer to tables in statements that declare a context without any special considerations, with the exception of on-action statements (latter must refer to the same context associated with the table).

You may also create a table and declare a context for the table. In this case the engine in effect manages separate tables, one for each context partition. Limitations apply in this case that we discuss herein.

For example, consider the 9 am to 5 pm fixed temoral context as shown earlier:

create context NineToFive start (0, 9, *, *, *) end (0, 17, *, *, *)

You may create a table that only exists between 9 am and 5 pm:

context NineToFive create table AverageSpeedTable (
  carId string primary key, 
  avgSpeed avg(double))

You can aggregate-into the table only if the aggregating statement declares the same context:

// declare the same context as for the table
context NineToFive into table AverageSpeedTable
select avg(speed) as avgSpeed
from TrafficEvent
group by carId

When you declare a context for a table, any select, on-merge, on-select, on-update and on-delete statements as well as statements that subquery the table must declare the same context.

For example, this EPL truncates the AverageSpeedTable:

context NineToFive on TruncateEvent delete from AverageSpeedTable

A variable is a scalar, object or event value that is available for use in all statements. Variables can be either global variables or context variables.

The value of a global variable is the same for all context partitions. The next example declares a global threshold variable:

create variable integer var_global_threshold = 100

For context variables, there is a variable value per context partition. The next example declares a context and a context variable:

create context ParkingLotContext initiated by CarArrivalEvent as cae terminated by CarDepartureEvent(lot = cae.lot)

context ParkingLotContext create variable integer var_parkinglot_threshold = 100

The variable var_parkinglot_threshold is a context variable. Each context partition can have its own value for the variable.

For more information on variables, please refer to Section 5.17, “Variables and Constants”.

Context variables can only be used in statements that associated to the same context.

Selecting specific context partitions and interrogating context partitions is useful for:

Esper provides APIs to identify, filter and select context partitions for statement iteration and on-demand queries. The APIs are described in detail at Section 16.18, “Context Partition Selection”.

For statement iteration, your application can provide context selector objects to the iterate and safeIterate methods on EPStatement. If your code does not provide context selectors the iteration considers all context partitions. At the time of iteration, the engine obtains the current set of context partitions and iterates each independently. If your statement has an order-by clause, the order-by clause orders within the context partition and does not order across context partitions.

For on-demand queries, your application can provide context selector objects to the executeQuery method on EPRuntime and to the execute method on EPOnDemandPreparedQuery. If your code does not provide context selectors the on-demand query considers all context partitions. At the time of on-demand query execution, the engine obtains the current set of context partitions and queries each independently. If the on-demand query has an order-by clause, the order-by clause orders within the context partition and does not order across context partitions.

The Event Processing Language (EPL) is a SQL-standard language with extensions, offering SELECT, FROM, WHERE, GROUP BY, HAVING and ORDER BY clauses. Streams replace tables as the source of data with events replacing rows as the basic unit of data. Since events are composed of data, the SQL concepts of correlation through joins, filtering and aggregation through grouping can be effectively leveraged.

The INSERT INTO clause is recast as a means of forwarding events to other streams for further downstream processing. External data accessible through JDBC may be queried and joined with the stream data. Additional clauses such as the PATTERN and OUTPUT clauses are also available to provide the missing SQL language constructs specific to event processing.

The purpose of the UPDATE clause is to update event properties. Update takes place before an event applies to any selecting statements or pattern statements.

EPL 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 EPL syntax. It also outlines the built-in views, which are the building blocks for deriving and aggregating information from event streams.

EPL statements contain definitions of one or more views. Similar to tables in a 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 EPL statements for efficiency.

The built-in set of views is:

EPL provides the concept of named window. Named windows are data windows that can be inserted-into and deleted-from by one or more statements, and that can queried by one or more statements. Named windows have a global character, being visible and shared across an engine instance beyond a single statement. Use the CREATE WINDOW clause to create named windows. Use the ON MERGE clause to atomically merge events into named window state, the INSERT INTO clause to insert data into a named window, the ON DELETE clause to remove events from a named window, the ON UPDATE clause to update events held by a named window and the ON SELECT clause to perform a query triggered by a pattern or arriving event on a named window. Finally, the name of the named window can occur in a statement's FROM clause to query a named window or include the named window in a join or subquery.

EPL provides the concept of table. Tables are globally-visible data structures that typically have primary key columns and that can hold aggregation state. You can create tables using CREATE TABLE. An overview of named windows and tables, and a comparison between them, can be found at Section 6.1, “Overview”. The aforementioned ON SELECT/MERGE/UPDATE/INSERT/DELETE, INSERT INTO as well as joins and subqueries can be used with tables as well.

EPL allows execution of on-demand (fire-and-forget, non-continuous, triggered by API) queries against named windows and tables through the runtime API. The query engine automatically indexes named window data for fast access by ON SELECT/MERGE/UPDATE/INSERT/DELETE without the need to create an index explicitly, or can access explicit (secondary) table indexes for operations on tables. For fast on-demand query execution via runtime API use the CREATE INDEX syntax to create an explicit index for the named window or table in question.

Use CREATE SCHEMA to declare an event type.

Variables can come in handy to parameterize statements and change parameters on-the-fly and in response to events. Variables can be used in an expression anywhere in a statement as well as in the output clause for dynamic control of output rates.

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

EPL queries are created and stored in the engine, and publish results to listeners as events are received by the engine or timer events occur that match the criteria specified in the query. Events can also be obtained from running EPL queries via the safeIterator and iterator methods that provide a pull-data API.

The select clause in an EPL query specifies the event properties or events to retrieve. The from clause in an EPL query specifies the event stream definitions and stream names to use. The where clause in an EPL 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#time(30 sec) where symbol='IBM'

EPL queries follow the below syntax. EPL queries can be simple queries or more complex queries. A simple select contains only a select clause and a single stream definition. Complex EPL 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.

[annotations]
[expression_declarations]
[context context_name]
[into table table_name]
[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]
[limit num_rows]

time-period : [year-part] [month-part] [week-part] [day-part] [hour-part] 
      [minute-part] [seconds-part] [milliseconds-part] [microseconds-part]

year-part : (number|variable_name) ("years" | "year")
month-part : (number|variable_name) ("months" | "month")
week-part : (number|variable_name) ("weeks" | "week")
day-part : (number|variable_name) ("days" | "day")
hour-part : (number|variable_name) ("hours" | "hour")
minute-part : (number|variable_name) ("minutes" | "minute" | "min")
seconds-part : (number|variable_name) ("seconds" | "second" | "sec")
milliseconds-part : (number|variable_name) ("milliseconds" | "millisecond" | "msec")
microseconds-part : (number|variable_name) ("microseconds" | "microsecond" | "usec")

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

// This comment extends to the end of the line.
// Two forward slashes with no whitespace between them begin such comments.

select * from MyEvent  // this is a slash-slash comment

// All of this text together is a valid statement.

/* This comment is a "slash-star" comment that spans multiple lines.
 * It begins with the slash-star sequence with no space between the '/' and '*' characters.
 * By convention, subsequent lines can begin with a star and are aligned, but this is 
 * not required.
 */		
select * from MyEvent  /* this also works */

select field1, // first comment
  /* second comment*/  field2
  from MyEvent

// valid
select last, count(*) as count from MyEvent

// invalid
select insert from MyEvent

// valid
select `insert` from `Order`

select * from OrderEvent(name='John\'s')
// ...equivalent to...
select * from OrderEvent(name='John\u0027s')

select * from OrderEvent(description like "Quote \"Hello\"")
// is equivalent to
select * from OrderEvent(description like "Quote \u0022Hello\u0022")

epService.getEPAdministrator().createEPL("select * from OrderEvent(name='John\\'s')");
// ... and for double quotes...
epService.getEPAdministrator().createEPL("select * from OrderEvent(
  description like \"Quote \\\"Hello\\\"\")");

select 'volume' as field1,
   "sleep" as field2, 
  "\u0041" as unicodeA

select true as field1, 
  false as field2

select 1 as field1, 
  -1 as field2, 
  1e2 as field3

select 1L as field1, 
  1l as field2

select 1.67 as field1, 
  167e-2 as field2, 
  1.67d as field3

select 1.2f as field1, 
  1.2F as field2

select 0x10 as field1

select * from MyEvent where property = MyConstantClass.FIELD_VALUE

select * from MyEvent where enumProp = EnumClass.ENUM_VALUE_1

select * from MyEvent where somevalue = EnumClass.ENUM_VALUE_1.getSomeValue()
  or EnumClass.ENUM_VALUE_2.analyze(someothervalue)

select * from MyEvent where enumProp = EnumClass.valueOf('ENUM_VALUE_1')

select * from MyEvent(enumProp=org.myorg.MyEvent$Color.GREEN)#firstevent

select myevent.computeSomething() as result from MyEvent as myevent

select myevent.getComputerFor('books', 'movies').calculate() as result 
from MyEvent as myevent

@annotation_name [(annotation_parameters)]

@annotation_name (attribute_name = attribute_value, [name=value, ...])

public @interface ProcessMonitor {
  String processName();
  boolean isLongRunning default false;
  int[] subProcessIds;
}

@ProcessMonitor(processName='CreditApproval',
  isLongRunning=true, subProcessIds = {1, 2, 3} )
  
select count(*) from ProcessEvent(processId in (1, 2, 3)#time(30)

epService.getEPAdministrator().getConfiguration().addAnnotationImport("com.mycompany.app.myannotations.*");

epService.getEPAdministrator().getConfiguration().addAnnotationImport("com.mycompany.myannotations.ProcessMonitor");

public enum MyEnum {
    ENUM_VALUE_1,
    ENUM_VALUE_2;
}

public @interface MyAnnotationWithEnum {
    MyEnum myEnum();
}

@MyAnnotationWithEnum(myEnum = enum_value_1) select * from MyEvent

@Name("MyStatementName")

@Description("Place statement
description here.")

@Tag(name="MyTagName", 
 value="MyTagValue")

@Priority(10)

@Drop

@Hint('iterate_only')

@Hook(type=HookType.SQLCOL,
  hook='MyDBTypeConvertor')

@Audit

@EventRepresentation(objectarray)

@EventRepresentation(avro)

@IterableUnbound

@Name("RevenuePerCustomer")
@Description("Outputs revenue per customer considering all events encountered so far.")
@Tag(name="grouping", value="customer")

select customerId, sum(revenue) from CustomerRevenueEvent

@Name("SecurityFilter1") select * from SecurityFilter(ip="127.0.0.1")

@Description('This statement filters localhost.') select * from SecurityFilter(ip="127.0.0.1")

@Tag(name='ip_sensitive', value='Y') 
@Tag(name='author', value='Jim')
select * from SecurityFilter(ip="127.0.0.1")

@Priority(10) select * from SecurityFilter(ip="127.0.0.1")

@Drop select * from SecurityFilter(ip="127.0.0.1")

@Hint('disable_reclaim_group') 
select ipaddress, count(*) from SecurityFilter#time(60 sec) group by ipaddress

expression expression_name alias for { expression }

expression twoPI alias for { Math.PI * 2 }
select twoPI from SampleEvent

expression countPeople alias for { count(*) }
select countPeople from EnterRoomEvent#time(10 seconds) having countPeople > 10

expression expression_name { expression_body }

expression_body:   input_param_name => expression 

expression_body:   (input_param [,input_param [,...]]) => expression 

expression twoPI { Math.PI * 2} select twoPI() from SampleEvent

expression twoPI { Math.PI * 2} select twoPI from SampleEvent

expression midPrice { x => (x.buy + x.sell) / 2 } 
select midPrice(md) from MarketDataEvent as md

expression midPrice { x => (buy + sell) / 2 } 
select midPrice(md) from MarketDataEvent as md

expression midPrice { x => (buy + sell) / 2 } 
select midPrice(*) from MarketDataEvent

expression weightedSentiment { (x, y) => x.price * y.sentiment } 
select weightedSentiment(md, news) 
from MarketDataEvent#lastevent as md, NewsEvent#lastevent news

expression newsSubq { md -> 
    (select sentiment from NewsEvent#unique(symbol) where symbol = md.symbol) 
} 
select newsSubq(mdstream)
from MarketDataEvent mdstream

... context context_name ...

The select clause also offers optional istream, irstream and rstream keywords to control whether input stream, remove stream or input and remove stream events are posted to UpdateListener instances and observers to a statement. By default, the engine provides only the insert stream to listener and observers. See Section 17.4.20, “Engine Settings related to Stream Selection” on how to change the default.

The syntax for the select clause is summarized below.

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

The istream keyword is the default, and indicates that the engine only delivers insert stream events to listeners and observers. The irstream keyword indicates that the engine delivers both insert and remove stream. Finally, the rstream keyword tells the engine to deliver only the remove stream.

The distinct keyword outputs only unique rows depending on the column list you have specified after it. It must occur after the select and after the optional stream keywords, as described in more detail below.

select * from stream_def

select * from StockTick(symbol='IBM')#time(30 sec)

select *, price * volume as pricevolume from StockTick

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

select symbol, price, sum(volume) from StockTick(symbol='IBM')#time(60 sec)

select datapoints, total, average, variance, stddev, stddevpa
from StockTick(symbol='IBM')#length(100)#uni(volume)