Flink CEP

Aug 04, 2019

A source-code-oriented introduction to Flink CEP, revisiting my earlier notes while preparing an online walkthrough.

Translated from Chinese with AI · Read the original

My earlier CEP articles are available here.

Those articles were fairly rough. While preparing an online session on the CEP source code, I decided to write down my understanding from an implementation perspective.

OnBoarding

As enterprise analytics becomes more sophisticated, event analysis is moving beyond offline computation and simple counts toward more timely, precise, and complex analysis. Complex event processing is complex because events are related, and those relationships take many temporal forms.

Let us begin with a test from CEPITCase.java. I have simplified testSimplePatternCEP():

1
@Test
2
public void testSimplePatternCEP() throws Exception {
3
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
4

5
  DataStream<Event> input = env.fromElements(
6
    new Event(1, "barfoo", 1.0),
7
    new Event(2, "start", 2.0),
8
    new Event(3, "foobar", 3.0),
9
    new SubEvent(4, "foo", 4.0, 1.0),
10
    new Event(5, "middle", 5.0),
11
    new SubEvent(6, "middle", 6.0, 2.0),
12
    new SubEvent(7, "bar", 3.0, 3.0),
13
    new Event(42, "42", 42.0),
14
    new Event(8, "end", 1.0)
15
  );
16

17
  Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
18

19
    @Override
20
    public boolean filter(Event value) throws Exception {
21
      return value.getName().equals("start");
22
    }
23
  }).followedBy("middle").subtype(SubEvent.class).where(
24
      new SimpleCondition<SubEvent>() {
25

26
        @Override
27
        public boolean filter(SubEvent value) throws Exception {
28
          return value.getName().equals("middle");
29
        }
30
      }
31
    )
32
  .followedBy("end").where(new SimpleCondition<Event>() {
33

34
    @Override
35
    public boolean filter(Event value) throws Exception {
36
      return value.getName().equals("end");
37
    }
38
  });
39

40
  DataStream<String> result = CEP.pattern(input, pattern).flatSelect((p, o) -> {
41
    StringBuilder builder = new StringBuilder();
42

43
    builder.append(p.get("start").get(0).getId()).append(",")
44
      .append(p.get("middle").get(0).getId()).append(",")
45
      .append(p.get("end").get(0).getId());
46

47
    o.collect(builder.toString());
48
  }, Types.STRING);
49

50
  List<String> resultList = new ArrayList<>();
51

52
  DataStreamUtils.collect(result).forEachRemaining(resultList::add);
53

54
  assertEquals(Arrays.asList("2,6,8"), resultList);
55
}

The pattern is fairly intuitive: find a sequence whose names are start, middle, and end. With this basic usage in mind, we can unpack CEP step by step.

Parsing Rules

Before considering Flink’s implementation, we can draw the relationships implied by this pattern. Data can be in four states:

The start condition has not been met.
start has been met, but middle has not.
start and middle have been met, but end has not.
start, middle, and end have all been met.

We name these states start, middle, end, and Final State. Their relationships are shown below.

cep-rule1

This makes the rule easier to understand. It also introduces the theoretical basis of flink-cep: the NFA, or nondeterministic finite automaton, in Efficient Pattern Matching over Event Streams. Nondeterministic means that the next state is not uniquely determined. Flink implements most of the paper’s core ideas, though features such as matching multiple rules are still missing. Let us examine how the source code represents an NFA.

cep-nfa-uml

Flink uses NFA.class for the model, State.class for states such as start, middle, and end, and StateTransition.class for their relationships. A StateTransition corresponds to an edge in the diagram: sourceState and targetState identify its endpoints, and condition determines when the transition occurs. Transitions have three possible actions:

1
public enum StateTransitionAction {
2
  TAKE, // 获取当前满足条件的事件
3
  IGNORE, // 忽略当前满足条件的事件
4
  PROCEED // 自然转换
5
}

PROCEED may be less intuitive; I will explain it later with an optional example.

We now have a basic understanding of NFAs and their representation in Flink. The UML diagram shows that State.class already contains its StateTransitions, so let us examine how these states and transitions are built. The source has many if/else branches for special cases; here we will focus on the example in the onboarding section.

The Pattern objects are nested, with the outermost object representing the final state. Unwrapping them therefore constructs the NFA in reverse order.

1
/**
2
 * Creates all the states between Start and Final state.
3
 *
4
 * @param sinkState the state that last state should point to (always the Final state)
5
 * @return the next state after Start in the resulting graph
6
 */
7
private State<T> createMiddleStates(final State<T> sinkState) {
8
  State<T> lastSink = sinkState;
9
  while (currentPattern.getPrevious() != null) {
10

11
    if (currentPattern.getQuantifier().getConsumingStrategy() == Quantifier.ConsumingStrategy.NOT_FOLLOW) {
12
      //skip notFollow patterns, they are converted into edge conditions
13
    } else if (currentPattern.getQuantifier().getConsumingStrategy() == Quantifier.ConsumingStrategy.NOT_NEXT) {
14
      省略 ................
15
    } else {
16
      lastSink = convertPattern(lastSink);
17
    }
18

19
    // we traverse the pattern graph backwards
20
    followingPattern = currentPattern;
21
    currentPattern = currentPattern.getPrevious();
22

23
    省略 ...........
24
  }
25
  return lastSink;
26
}

Indeed, the code recursively unwraps Pattern and calls convertPattern to parse each part. Ignore the special cases for NOT_FOLLOW and NOT_NEXT for now. convertPattern calls the following function:

1
private State<T> createSingletonState(final State<T> sinkState,
2
      final State<T> proceedState,
3
      final IterativeCondition<T> takeCondition,
4
      final IterativeCondition<T> ignoreCondition,
5
      final boolean isOptional)

It first derives possible next states and their conditions from currentPattern, including its where condition and ConsumingStrategy, then calls createSingletonState to create the state. The example therefore produces the following NFA:

flink-nfa-1

We can now describe more complex cases, such as optional and times.

// TODO

Matching Rules

// TODO

Producing Output

// TODO

Things to Watch When Using CEP

// TODO