CEP In Flink (2) - Matching CEP Rules

Feb 14, 2019

How Apache Flink matches CEP rules.

Translated from Chinese with AI · Read the original

The previous post explained that Flink’s implementation follows this paper. Let’s first examine the data structures built on that theory.

Data Structures

ShareBuffer

After parsing, a CEP rule is essentially a nondeterministic state machine. During matching, each state corresponds to one or more elements. Consider three states, omitting their where conditions:

1
Pattern.begin("A").followedBy("B").followedBy("C")

The diagram is: cep2-1

This structure is called SharedBuffer. Each event is represented by a NodeId composed of an EventId and pageName, the state name. Each NodeId maps to a SharedBufferNode containing edges to preceding events. Reference counting controls its lifetime. The reason for this structure is discussed below.

NFAState

Another important structure is NFAState, which retains the current matching state. Its two key variables are:

1
private Queue<ComputationState> partialMatches; // 正在进行的匹配
2
private Queue<ComputationState> completedMatches; // 完成的匹配

For each new event, partialMatches is traversed to see whether any partial match can become a completed match.

Matching Process

Given these structures, the process is fairly predictable. On receiving an event, traverse partialMatches, compute each match’s next state, and add the current event to SharedBuffer. Reaching a final state turns a partial match into a completed match.

1
final PriorityQueue<ComputationState> newPartialMatches = new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR);
2
final PriorityQueue<ComputationState> potentialMatches = new PriorityQueue<>(NFAState.COMPUTATION_STATE_COMPARATOR);
3

4
// iterate over all current computations
5
for (ComputationState computationState : nfaState.getPartialMatches()) {
6
    final Collection<ComputationState> newComputationStates = computeNextStates(
7
        sharedBuffer,
8
        computationState,
9
        event,
10
        event.getTimestamp());
11

12
    if (newComputationStates.size() != 1) {
13
        nfaState.setStateChanged();
14
    } else if (!newComputationStates.iterator().next().equals(computationState)) {
15
        nfaState.setStateChanged();
16
    }
17

18
    //delay adding new computation states in case a stop state is reached and we discard the path.
19
    final Collection<ComputationState> statesToRetain = new ArrayList<>();
20
    //if stop state reached in this path
21
    boolean shouldDiscardPath = false;
22
    for (final ComputationState newComputationState : newComputationStates) {
23

24
        if (isFinalState(newComputationState)) {
25
            potentialMatches.add(newComputationState);
26
        } else if (isStopState(newComputationState)) {
27
            //reached stop state. release entry for the stop state
28
            shouldDiscardPath = true;
29
            sharedBuffer.releaseNode(newComputationState.getPreviousBufferEntry());
30
        } else {
31
            // add new computation state; it will be processed once the next event arrives
32
            statesToRetain.add(newComputationState);
33
        }
34
    }
35

36
    if (shouldDiscardPath) {
37
        // a stop state was reached in this branch. release branch which results in removing previous event from
38
        // the buffer
39
        for (final ComputationState state : statesToRetain) {
40
            sharedBuffer.releaseNode(state.getPreviousBufferEntry());
41
        }
42
    } else {
43
        newPartialMatches.addAll(statesToRetain);
44
    }
45
}

In NFA.java, computeNextStates determines the next states for the current partial match. isFinalState and isStopState then classify and handle them, while nodes in SharedBuffer are released.

We can simplify computeNextStates to the following:

1
final OutgoingEdges<T> outgoingEdges = createDecisionGraph(context, computationState, event.getEvent())
2
final List<StateTransition<T>> edges = outgoingEdges.getEdges();
3
final List<ComputationState> resultingComputationStates = new ArrayList<>();
4
for (StateTransition<T> edge : edges) {
5
    switch (edge.getAction()) {
6
        case IGNORE:
7
            // ...处理ignore
8
            break;
9
        case TAKE:
10
            // ...处理take
11
            break;
12
    }
13
}

createDecisionGraph evaluates the current event against the transition graph described in CEP In Flink (1), producing eligible outgoingEdges. Each edge has an action and targetState. Traversing them creates new states returned to the previous code fragment.

Matching Optimization

As events arrive, more partial matches coexist. To avoid duplicating events, SharedBuffer gives each SharedBufferNode versioned SharedBufferEdges pointing to different preceding NodeIds, as shown:

CEP diagram

Events arrive in the order e0 -> e1 -> e2 -> e3. Here, e3 follows both e1 and e2, distinguished by version numbers. Reality is slightly more complex, for example version 1.0 is omitted from the diagram, but the overall optimization follows the same idea.