Understanding Flink CEP's NFAb Automaton for Complex Event Processing

Flink's Complex Event Processing (CEP) library enables detection of event correlations in unbounded data streams by matching predefined patterns, supporting use cases such as trend analysis, risk monitoring, and fraud detection. It offers a concise, expressive API, exemplified by a Scala snippet that sets event time, defines a pattern, and selects alerts.

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
val partitionedInput = sourceStream.keyBy(event => event.getId)
// start[] -> middle[name = 'error'] -> .. -> end[name = 'critical'] within 10 secs
val pattern = Pattern.begin[Event]("start")
  .next("middle").where(_ .getName == "error")
  .followedBy("end").where(_ .getName == "critical")
  .within(Time.seconds(10))
val patternStream = CEP.pattern(partitionedInput, pattern)
val alerts = patternStream.select(createAlert(_))

The underlying matching algorithm is based on the paper "Efficient Pattern Matching over Event Streams" and uses a nondeterministic finite automaton with matching caches (NFAb). The NFAb is defined as a five‑tuple (Q, E, θ, q₁, F) where Q is the set of states, E the directed edges, θ the transition predicates, q₁ the start state, and F the final states.

An example pattern described in the SASE+ language shows a stock‑trend detection rule with a one‑hour window, requiring the first event's volume > 1000, subsequent prices higher than previous averages, and a final drop below 80% of the previous volume.

The article illustrates how this pattern is compiled into an NFAb automaton, with diagrams of the automaton and its transitions. Four transition semantics are defined: begin, take, ignore, and proceed; Flink CEP implements take, ignore, and proceed, which are equivalent to the paper's semantics.

Event selection strategies are discussed: strict (next/notNext), skip‑till‑next‑match (followedBy/notFollowedBy), and skip‑till‑any‑match (followedByAny). The skip‑till‑next‑match strategy is demonstrated with example data that yields three matching sequences (R1, R2, R3).

To avoid exponential growth of match caches, the paper proposes a shared versioned match buffer that attaches version numbers to forward pointers, allowing efficient reconstruction of sequences. Flink CEP implements a similar structure in the SharedBuffer class.

Each matching sequence also maintains a computation state containing the current version, state, pointer to the latest cached event, start time, and auxiliary context such as event counts, price sums, and volumes. Flink CEP uses the ComputationState class for this purpose.

© The article is authored by the "import_bigdata" team and is intended for readers interested in big‑data stream processing and CEP techniques.