CEP In Flink (1) - Parsing CEP Rules

Introduction

As data analysis becomes more detailed, enterprise event analysis is moving beyond ordinary batch computation and simple counts toward more immediate, precise, and complex analysis. Complex event processing is complex because events are related, and their temporal relationships take many forms. An earlier post, Oracle CEP Syntax, gave an example, so I will not repeat it here.

Current Landscape

• EsperTech developed its own DSL for describing complex events and uses it in its enterprise products.
• Siddhi is an open-source framework for stream processing and complex event analysis, used by Uber and Apache Eagle. It is lightweight enough to deploy on user devices.
• Oracle and Calcite both provide relatively complete SQL syntax for complex events; see Oracle CEP Syntax.

Complex event relationships lead to complex description languages. The three services above use different languages and APIs with relatively steep learning curves. Calcite and Oracle can both express CEP relationships in SQL, and Flink CEP is implementing SQL support based on Calcite. I think standardized SQL makes this easier to understand and adopt widely.

Theoretical Foundation

Apache Flink’s CEP implementation draws on the NFA model in Efficient Pattern Matching over Event Streams. The paper also describes optimizations, which we will skip for now to focus on rule parsing.

The paper introduces the NFA, or nondeterministic finite automaton: it has a finite number of states, but each state may transition to several possible states, hence the nondeterminism.

Let’s use a simple CEP rule to see how the NFA represents event relationships.

1
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
2
  @Override
3
  public boolean filter(Event value) throws Exception {
4
    return value.getName().equals("a");
5
  }
6
}).followedBy("middle").optional().where(new SimpleCondition<Event>() {
7
  @Override
8
  public boolean filter(Event value) throws Exception {
9
    return value.getName().equals("b");
10
  }
11
}).followedBy("end").where(new SimpleCondition<Event>() {
12
  @Override
13
  public boolean filter(Event value) throws Exception {
14
    return value.getName().equals("c");
15
  }
16
});

The code defines the relationships and conditions for start, middle, and end. Middle follows start, and end follows middle, but strict adjacency is not required. Middle is optional. The NFA representation is shown below:

First, two concepts:

• States: start/middle/middle/end/$end$ are state names.
• Transitions: take/ignore/proceed are transition names.

Matching an NFA rule is fundamentally a process of state transitions. The three transitions mean:

• Take: The condition is met; consume the current element and enter the next state.
• Proceed: Without consuming the current element, enter the next state regardless of whether the current condition holds, as with optional, and test its condition.
• Ignore: The condition is not met; ignore the element and enter the next state.

Now consider a stream containing only four elements:

start -> xx -> middle -> end

1. Receive start: the condition holds, so take transitions to middle.
2. Receive xx: the condition fails, so the state becomes middle. Proceed also leads to end, whose condition fails too, so that transition is discarded.
3. Receive middle: middle satisfies the condition and takes a transition to end. At end, the condition fails and ignore transitions back to itself.
4. Receive end: the condition holds, so take transitions to $end$ and completes the match.

Summary

This first post in the Flink CEP series introduces background concepts and uses a simple example to explain how an NFA represents CEP rules. The next post will examine the matching process itself. Stay tuned.