How To's / Common Design Patterns

You generally want to only keep a small portion of the data in Complex Event Processing (CEP) applications. In the Aleri Streaming Platform, windows are used to automatically purge data. But occasionally, you might need to keep track of statistics for all of the events, not just those in the window. How do you solve this "maintaining statistics" problem?

The Aleri Streaming Platform has a dictionary data structure that can be used to solve this common problem.

A concrete example of this problem would be keeping track of the minimum and maximum price of stock trades. The data model (with associated test data), has a source stream "Trades" with fields

The FlexStream keeps two local dictionaries with declarations

This method can be extended to other statistics of interest, such as keeping the average price over all events.

The Aleri Streaming Platform allows you to manipulate structured XML data like most databases. A value of "xml" type is a tree structure like

The operations are a subset of the SQL/XML standard. You create variables holding xml values and assign them to values such as

and convert them back and forth to strings.

Here's an example model and small data set that demonstrates one use of the xml data type: creating joins that propagate changes in an atomic way. It takes in events with fields

and events with fields

outputs structures like

If you look at the model in the Aleri Studio, you'll see a diagram

with two source streams "departments" and "members", a Join Stream "fullDept" with fields

an Aggregate Stream "groupDept" with fields

and a similar Aggregate Stream "groupMgr" stream. These aggregations make the changes from the Join Stream atomic. The above structure will be altered whenever a new department is added or a member of the department changes. Those changes are harder to see from the Join Stream alone. For example, if a member of a department changes, the Join Stream will only output the changed row instead of the full department.

It's instructive to look at the computation inside "groupDept". The expression for computing the above structure is

The important operation is the "xmlagg" function, which aggregates a number of xml values into one xml value. More precisely, each "member" element from the "fullDept" stream is copied into the structure, which is wrapped in a "department" tag at the top level.

Future releases will have more functionality added to the xml data type, such as searching and replacing elements using XPath expressions.

This is the first of the design patterns considered in this document, where multiple events are kept in memory. In-memory data caching is the foundation of most CEP design patterns. The cache typically stores two kinds of data:

• Recent events from one or more streams
  Recent events are typically stored in windows. A window is an object, similar to an in-memory database table. However, a window can manage its state automatically, by keeping and evicting certain events according to its policy. For example a window policy might specify: KEEP 1000 ROWS PER Id. This window maintains 1000 rows for each ID value, and expires old rows, as necessary.
• Data from one or more database tables
  Just as streaming events can be cached in memory, it often makes sense to cache data from a relational database, so that different kinds of operations may be performed on this data more efficiently. This cache is typically managed according to the Least Recently Used (LRU) algorithm, or by explicit invalidation.

(Click Image to Enlarge)

Note that, although we are describing an in-memory cache here, many applications require this cache to be persistent. This means that, if a machine that hosts the CEP engine fails, the data kept in windows is not lost. This functionality is even more important when the window holds not just the last few seconds’, but minutes’, hours’, days’, and even weeks’ worth of events.

Sample Areas of Applicability

In-memory caching is used in every non-trivial CEP application. Here are just a few examples:

• Trading: in a trading application, the cache may hold the values of recent trades, recent orders, or recent news events, coupled with the relevant historical and reference information.
• Click-stream analysis: a typical application may hold the recent clicks and searches performed by the users, coupled with the relevant historical and reference information.
• Network security: A typical application may hold recent events from firewalls, intrusion detection systems, and other devices, coupled with the relevant historical and reference information.

CCL Examples

In this example, we will focus on the first form of caching, namely, caching events from a data stream. The Database Lookup design pattern, presented later in this document, discusses the caching of database data.

Windows

Looking up Data in a Window

While there are applications that deal exclusively with real-time events, most useful applications refer to historical data or reference data to enrich the incoming events. The following diagram shows how:

• An event comes into the system.
• The engine issues an SQL request to the database and passes a key (from the event) as a parameter to the database query.
• The database returns a result.
• The engine combines the result with data from the event, and forwards the enriched event to the next query for further processing.

(Click Image to View Larger)

In SQL terminology, this design pattern implements a join between an incoming data stream and a database table.

The more advanced implementations of this design pattern often require:

• Caching
  A high throughput is difficult to achieve if the CEP engine must refer to the database on every incoming event. Thus, a good caching layer is critical to performance.
• Granular caching and access
  Sometimes it is acceptable to cache entire database tables in memory. Such a paradigm works for small database tables, but does not scale well if many large tables need to be accessed. For example, if an RFID application is trying to look up information about a specific RFID tag, caching the entire database table is often not possible. Accessing and caching specific rows is required for these applications.
• Concurrent lookups
  For some applications, it is acceptable to issue serial calls to the database, blocking the processing of events until the database returns results. For others, concurrent lookups that do not block the entire system are a must.

Sample Areas of Applicability

This design pattern is widely applicable. For example:

• Trading: a trading application may look up historical price for a stock, or certain information about an order, or certain rules and regulations stored in a database.
• RFID application: an application may look up information about a palette or case, identified by its tag ID, or information about the reader that reported the tag. An application may also check where the object should be located, according to the plan stored in the database, and compare this location to the actual location of the object.
• Network security: when deciding how serious an alert is, it may be necessary to refer to other alerts related to the same IP address.

CCL Examples

In these examples, the SQL lookup code (delimited by [[ ... ]]) is executed in the database, not in the Coral8 Engine.


-- Look up the SKU and the name of the product for each RFID event, via the Tag ID.
INSERT INTO
StreamRFIDEventsEnriched
SELECT
StreamRFIDEvents.TagID, DbResult.SKU, DbResult.ProductName
FROM StreamRFIDEvents,
(DATABASE "OracleDb"
SCHEMA "schemas/product_info.ccs"
-- SQL Code to perform lookup:
[[SELECT sku, product_name
FROM product_info pi
WHERE pi.tag_id = StreamRFIDEvents.TagID]]) AS DbResult;

-- Look up yesterday’s closing prices for each incoming trade.

INSERT INTO
StreamTradesEnriched
SELECT
InTrades.Symbol, InTrades.Price, DbResult.ClosingPrice
FROM StreamTrades,
(DATABASE "OracleDb"
SCHEMA "schemas/closing-price.ccs"
-- SQL Code to perform lookup:
[[SELECT closing_price
FROM price_history ph
WHERE ph.symbol = ?InTrades.Symbol
AND ph.closing_date > current_date-2
AND ph.closing_date <= current_date-1]]
) as DbResult