Complex Event Processing

NebulaStreamCEP presents the experimental Complex Event Processing (CEP) feature of NebulaStream and enables users to detect specified patterns in event streams. It extends NebulaStream’s operator set with CEP-specific operators, i.e., conjunction, sequence, disjunction, and times. In the remainder, Figure 1 is used to support the explanation of these operators. Furthermore, the stream processing operators and especially the mandatory window operator can be used to specify patterns.

💡 For accurate pattern detection, the sliding window operator with a slide size equivalent to the ingestion rate of the most frequent stream as well as event time needs to be applied to the pattern. Other window types and slide sizes can be applied but may lead to the loss of matches.

Conjunction #

The conjunction operator is a binary operator that expects a record (event) from stream1 and a record from stream2 to occur within a window, regardless of their temporal order. In the following, we apply the conjunction operator andWith to two logical streams. The output schema contains the attributes of both schemas together with the window start and window end time.

• C++
• Java

Query::from("stream1")
.andWith(Query::from("stream2"))
.window(SlidingWindow::of(EventTime(Attribute("timestamp")),Minutes(5),Minutes(1)));

currently not supported

For the conjunction pattern above, the highlighted window of Figure 1 returns three matches, i.e., each record of stream2 combined with the record of stream1.

Sequence #

The sequence operator is a binary operator that expects a record from stream1 and a record from stream2 to occur within a window. In contrast to the conjunction operator, the sequence operator is a temporal operator that requires the records to occur in the specified temporal order, i.e., the timestamp of record from stream1 is smaller than the timestamp of the record from stream2. In the following, we apply the sequence operator seqWith to two logical streams. The output schema contains the attributes of both schemas together with the window start and window end time.

• C++
• Java

Query::from("stream1")
.seqWith(Query::from("stream2"))
.window(SlidingWindow::of(EventTime(Attribute("timestamp")),Minutes(5),Minutes(1)));

currently not supported

For the sequence pattern above, the highlighted window of Figure 1 returns two matches, i.e., the record of stream1 composed with each of the two subsequent records of stream2.

Disjunction #

💡 Currently, it is not possible to apply a window operator to the disjunction pattern.

The disjunction operator is a binary operator that expects either a record from stream1 or a record from stream2 to occur within a window, regardless of any temporal order. Furthermore, the schema of both data streams has to be union compatible. If necessary, we can use projections to make the streams union compatible. In the following, we apply the disjunction operator orWith to two logical streams.

• C++
• Java

Query::from("stream1")
.orWith(Query::from("stream3"))
.window(SlidingWindow::of(EventTime(Attribute("timestamp")),Minutes(5),Minutes(1)));

currently not supported

For the disjunction pattern above, the highlighted window of Figure 1 returns the record of stream1 as a match.

Times #

The times operator, also called iteration operator in CEP, is a unary operator that allows to detect multiple occurrences of a specific event type (stream) in a sequence. Currently, NebulaStreamCEP supports an approximate version of this operator that detects if a user-defined number of events occurred within a window and returns a single record per window with the number of events that have been assigned to this window. In particular, NebulaStreamCEP allows for the following variants:

• .times() The variant handles an unlimited amount of event occurrences but requires at least one record per window, i.e., one record of the specific stream has to occur.

• .times(n) This variant enables the specification of patterns that detect event sequences of n records in a window, where n > 0.

• .times(n,m) This variant enables the specification of patterns that detect event sequences of at least n and at most m records in a window. (n,m > 0)

  • 💡 Note:
    • if n = 0, the result is equivalent to .times(n)
    • if m = 0, the result is an unbounded variant of the times operator that expects at least n occurrences of the specified event.

In the following, we apply variant .times(n,m) to a logical stream.

• C++
• Java

Query::from("stream2")
.times(3,10)
.window(SlidingWindow::of(EventTime(Attribute("timestamp")),Minutes(5),Minutes(1)));

currently not supported

For the iteration pattern above, the highlighted window of Figure 1 returns one matches, i.e., a record of count 3.

Not #

💡 Not yet supported

Selection and Consumption Policies #

NebulaStreamCEP supports the selection policy skip-till-any-match and the consumption policy non-consuming. The support of more restrictive policies is planned.