CEP In Flink (3) - Extracting Matched Events

Feb 20, 2019

How matched events are extracted in Apache Flink’s CEP module.

Translated from Chinese with AI · Read the original

The previous post described matching and its optimizations. What happens to events after a match?

Reconstructing Event Sequences

When a state reaches Final, its match enters potentialMatches as a candidate for output. SharedBuffer traces backward from the final match, using version numbers and NodeIds, described in the previous post, to reconstruct a unique event path. Here is the code:

1
for (ComputationState match : potentialMatches) {
2
    // 因为要从EventId还原到具体的T类型事件，做缓存提高效率
3
  Map<EventId, T> eventsCache = new HashMap<>();
4

5
    // 从SharedBuffer中，根据match的上一个NodeId和版本不断向前提取事件
6
  Map<String, List<T>> materializedMatch =
7
    sharedBuffer.materializeMatch(sharedBuffer.extractPatterns(
8
        match.getPreviousBufferEntry(),
9
        match.getVersion()).get(0),
10
      eventsCache);
11

12
    // 释放已经匹配完成的Node
13
  sharedBuffer.releaseNode(match.getPreviousBufferEntry());
14
}

Here, Map<String, List<T>> represents the relationship between state names and their events.

In AbstractKeyedCEPPatternOperator.java, after nfa.process completes, the return type is Collection<Map<String, List<IN>>>. String is the state name and List<IN> contains the events matched by that state. The code is:

1
/**
2
 * Process the given event by giving it to the NFA and outputting the produced set of matched
3
 * event sequences.
4
 *
5
 * @param nfaState Our NFAState object
6
 * @param event The current event to be processed
7
 * @param timestamp The timestamp of the event
8
 */
9
private void processEvent(NFAState nfaState, IN event, long timestamp) throws Exception {
10
  Collection<Map<String, List<IN>>> patterns =
11
      nfa.process(partialMatches, nfaState, event, timestamp, afterMatchSkipStrategy);
12
  processMatchedSequences(patterns, timestamp);
13
}

The role of processMatchedSequences is explained next.

Event Output

Consider the test in CEPITCase.java:

1
CEP.pattern(input, pattern).select(
2
  new PatternSelectFunction<Event, String>() {
3

4
    @Override
5
    public String select(Map<String, List<Event>> pattern) {
6
      StringBuilder builder = new StringBuilder();
7
      // 对builder的一些操作
8
      return builder.toString();
9
    }
10
  }
11
);

processMatchedSequences directly invokes the user-defined extraction function. Flink supports several forms, including flatSelect and select.

AfterMatchSkipStrategy

Another interesting feature is AfterMatchSkipStrategy. Used well, it filters many invalid or unwanted matches.

Suppose you want to detect a W-shaped stock-price pattern:

CEP stock-price W pattern

Both acde and bcde satisfy the W pattern, but do you want both outputs? Not necessarily. AfterMatchSkipStrategy removes completed or ongoing matches according to configured pruning rules after a successful match. Its interface is:

1
/**
2
 * Prunes matches/partial matches based on the chosen strategy.
3
 *
4
 * @param matchesToPrune current partial matches
5
 * @param matchedResult  already completed matches
6
 * @param sharedBuffer   corresponding shared buffer
7
 * @throws Exception thrown if could not access the state
8
 */
9
public void prune(
10
    Collection<ComputationState> matchesToPrune,
11
    Collection<Map<String, List<EventId>>> matchedResult,
12
    SharedBuffer<?> sharedBuffer) throws Exception {
13
    // 挑选出开始prune的EventId（或者说位置）
14
  EventId pruningId = getPruningId(matchedResult);
15
  if (pruningId != null) {
16
    List<ComputationState> discardStates = new ArrayList<>();
17

18
    // 针对每一个已经匹配成功的match做检查
19
    for (ComputationState computationState : matchesToPrune) {
20
      if (computationState.getStartEventID() != null &&
21
        shouldPrune(computationState.getStartEventID(), pruningId)) {
22
        sharedBuffer.releaseNode(computationState.getPreviousBufferEntry());
23
        discardStates.add(computationState);
24
      }
25
    }
26
    matchesToPrune.removeAll(discardStates);
27
  }
28
}

Custom AfterMatchSkipStrategies implement getPruningId and shouldPrune.

1
/**
2
 * Tells if the partial/completed match starting at given id should be prunned by given pruningId.
3
 *
4
 * @param startEventID starting event id of a partial/completed match
5
 * @param pruningId   pruningId calculated by this strategy
6
 * @return true if the match should be pruned
7
 */
8
protected abstract boolean shouldPrune(EventId startEventID, EventId pruningId);
9

10
/**
11
 * Retrieves event id of the pruning element from the given match based on the strategy.
12
 *
13
 * @param match match corresponding to which should the pruning happen
14
 * @return pruning event id
15
 */
16
protected abstract EventId getPruningId(Collection<Map<String, List<EventId>>> match);

Common predefined strategies include: