Stream Processing (3) - Rule Engines and Flink CEP

Complex event processing, or CEP, is often called a rule engine in enterprise practice. As real-time warehouses mature, CEP will become another major direction for stream-processing teams.

Other articles in this series:

• Stream Processing (1) - Why Not RocksDB in Streaming State
• Stream Processing (2) - Flink Fault Tolerance and Weakly Consistent Snapshots
• Stream Processing (3) - A Rule System Based on Flink CEP

What Is CEP?

CEP stands for Complex Event Processing. Its complexity comes from combining computational patterns beyond ordinary real-time processing, rather than merely complicated business logic. Examples include:

• Temporal context: Detect a fraudulent sequence and ban users who perform one action, then another, then a third.
• Negation: Send a coupon to users who view a product but do not order within ten minutes.
• Custom statistical conditions: Escalate a microservice alert if another alert occurs within thirty seconds of the first.

CEP remains a form of real-time computation, so most users build on Flink or an existing service, often through a library. These are common business scenarios, and most engineers could implement a few individually.

General System Architecture

In practice, it takes more than a few SQL statements or lines of code. Rules are usually authored by nontechnical operations, merchant, or marketing staff, and results often reach users immediately as coupons or notifications.

Unlike traditional BI, this computation directly affects business outcomes, making the system more complex. Companies therefore often package it as a rule-engine service.

A typical architecture looks like this:

Implementation Challenges

Because output directly affects end users, rule engines require greater rigor than ordinary real-time warehouses:

Complex components: Inputs are diverse and may require windows, multistream joins, and other processing. Outputs need validation and safeguards. A usable platform must cover the complete campaign loop from rule configuration to ROI measurement.

Offline/online inconsistency: CEP is online and timely, but results depend strongly on event order. Even event-time processing may discard late events beyond watermarks. Reproducing temporal logic offline after a complaint is difficult, and a correct recomputation may still differ from the original run because of backlog or delayed client uploads.

Correctness checks: Coupon distribution and ad delivery feed ROI calculations, so every rule trigger needs validation and safeguards, such as frequency caps or maximum trigger counts per rule.

CEP in Flink

Flink CEP is a library, independent of the engine implementation and built on low-level APIs. Reading it reveals useful Flink techniques. Its main stages are:

• Rule parsing
• Rule matching
• Extracting matched events

Rule Parsing

Flink CEP uses the NFA model from Efficient Pattern Matching over Event Streams. The paper also discusses memory optimizations, which we will revisit later.

An NFA is a nondeterministic finite automaton: it has finitely many states, but a state can transition to more than one possible next state.

Consider a simple rule and its event relationships:

1
Pattern<Event, ?> pattern = Pattern.<Event>begin("begin").where(new SimpleCondition<Event>() {
2
    @Override
3
    public boolean filter(Event value) throws Exception {
4
        return value.getName().equals("a");
5
    }
6
}).followedBy("middle").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 rule looks for a->b->c. Its corresponding NFA is a state-transition graph:

Each node represents a matching stage. The begin node is the initial state, retained until data satisfying value="a" arrives. Edges encode transition conditions; for example, value="a" permits begin to middle. Edge actions can be classified as:

• TAKE: Consume the event and move to the next state.
• IGNORE: Ignore the event and retain the current state.
• PROCEED: Attempt matching from the next state, as with optional patterns.

Rule Matching

After parsing creates the NFA, incoming events drive matching. Intermediate progress must be stored. The NFA uses a shared buffer, implemented with Flink state for event details. For a->b->c with input a1,b1,c1, the match is a1->b1->c1:

Now consider a1,a2,b1,b2,c1. The operator produces four matches:

1. a1->b1->c1
2. a1->b2->c1
3. a2->b1->c1
4. a2->b2->c2

All four satisfy the rule. How does one NFA graph track several matches simultaneously? For each record, the pseudocode is:

1
for state in partialStates:   // 遍历正在匹配中的状态
2
    for edge in state.edges:   // 遍历状态的边，逐一检查是否满足条件
3
        if match:  // 如果满足，状态发生转移
4
            partialStates.remove(state)
5
            newState = state.transTo(edge.targetState)
6
            partialStates.add(newState)
7

8
// 如果初始化状态发生了转化，新增一个初始化状态，准备新的一次匹配
9
if not partialStates.contains(beginState):
10
    partialStates.add(beginState)

Sequences are not stored separately. Each state node has a list of events connected by backward pointers, allowing event storage to be shared. More on this during match extraction.

Business rules usually impose time limits; otherwise matching could continue forever. For example, event A followed by B within a day:

1
Pattern<Event, ?> pattern = Pattern.<Event>begin("begin").where(new SimpleCondition<Event>() {
2
    @Override
3
    public boolean filter(Event value) throws Exception {
4
        return value.getName().equals("a");
5
    }
6
}).followedByAny("middle").where(new SimpleCondition<Event>() {
7
    @Override
8
    public boolean filter(Event value) throws Exception {
9
        return value.getName().equals("b");
10
    }
11
}).within(Time.days(1));

within(Time) sets the sequence’s matching window. Unlike calendar-aligned windows, this starts with the first matching event: a match beginning at 18 expires at 18 the next day. CEP registers a timer and stores startTimestamp. When the timer fires, it scans active matches and applies user-defined timeout handling to those satisfying currentTime > startTimestamp + 1day.

Flink offers many CEP matching semantics, listed at https://nightlies.apache.org/flink/flink-docs-master/docs/libs/cep/. Its rich streaming APIs mean CEP does not require capabilities beyond the underlying engine.

Extracting Matched Events

After matching a->b->c reaches the output state, Flink must emit its event list through this user API:

1
class MyPatternProcessFunction<IN, OUT> extends PatternProcessFunction<IN, OUT> {
2
    @Override
3
    public void processMatch(Map<String, List<IN>> match, Context ctx, Collector<OUT> out) throws Exception;
4
        IN startEvent = match.get("start").get(0);
5
        IN endEvent = match.get("end").get(0);
6
        out.collect(OUT(startEvent, endEvent));
7
    }
8
}

Map<String, List<IN>> match represents one successful match. Map keys are state names, and lists contain the events for each state. With several simultaneous matches, how does Flink identify the events belonging to each output?

Events are shared through lists and backward pointers. To disambiguate paths, Flink assigns versions to edges, following the paper’s shared-buffer idea. Output extraction traces the appropriate versioned path:

For a->multiple b->c, edges connect related events and version prefixes determine compatibility: 1.0.0 matches 1.0, and 1.0.1.0 matches 1.0.1. On completion, walk backward from the last event. Versions depend on transition counts. When b2 arrives after b1 in middle, one path advances to end and another remains in middle to match more b events, producing versions 1.0.0 and 1.0.1.

Flink differs somewhat from the paper’s shared buffer, which considers multiple-rule matching more extensively:

In the paper, version length represents state-path length, and branch counts advance versions. At e5, a branch changes the e6->e5 edge from 1.0 to 1.1. Compatibility extends downward at the current path length, so 1.1 is compatible with 1.0. See the paper for full details.

Limitations

Flink’s NFA-based implementation is fairly complete, but real CEP applications are complex, and larger deployments often cannot use the open-source implementation directly:

• Detailed-event storage: NFA matching retains events at every state for the user API. Many applications only need an audience’s final user IDs. Retaining all details limits complex rules and large-volume workloads.
• No aggregation: A rule such as more than five alerts in thirty minutes is implemented simply with five state nodes for times(5). This does not generalize to conditions such as summed monetary amounts.
• Difficult debugging: User-facing errors need explanations and validation. Streaming replay and snapshot restoration already complicate offline verification; CEP’s matching logic adds another layer of difficulty.

Other CEP Engines

Siddhi is another capable option, positioned as an embedded streaming framework with its own syntax. Users must arrange distributed deployment, perhaps with Kubernetes, and maintain another stack alongside Flink. Hao Chen’s flink-siddhi project integrates the two.

Summary

This article explained rule-engine architecture and Flink CEP’s internals. As real-time warehouses spread, companies will move beyond BI reporting into more complex applications, with CEP an important use case.

A rule engine is an entire system. Common approaches combine Flink with custom CEP or business operators, or build custom logic on online services and storage. Either way, architects invest heavily in the complete end-to-end pipeline. Existing infrastructure and open-source projects remain insufficient; I hope more specialized, comprehensive systems emerge.