根据第三个事件将两个事件链接在Flink中

Question

如果两个事件流到flink中，是否可以使用跟随它们的第三个事件（使用dataStream API或CEP）在逻辑上加入它们？例如，下面示例中的第三个事件是否可以根据其right_id和left_id链接前两个事件？

ID: AAAA
ID: BBBB
ID: ZZZZ, right_id: AAAA, left_id: BBBB 

Answer 1

这是一个非常基本的CEP用例。代码看起来像这样...

// data stream creation 
DataStream<Event> myStream = ... 
// cep pattern definition
Pattern<Event, ?> myPattern = Pattern.<Event>begin("first_event")
                .followedBy("second_event")
                .followedBy("match_event");
// cep pattern stream: apply pattern to stream
PatternStream<Event> myPatternStream = CEP.pattern(myStream, myPattern);
// create new data stream from pattern matches
DataStream<CEPEvent> myCEPEvent = myPatternStream.flatSelect(
                (Map<String, Event> pattern, Collector<CEPEvent> out) -> {
                // load potential event sequence matches
                Event first_event = pattern.get("first_event");
                Event second_event = pattern.get("second_event");
                Event match_event = pattern.get("match_event");
                // test event sequences 
                if (match_event.right_id.equals(first_event.ID)
                    && match_event.left_id.equals(second_event.ID)
                ){out.collect(new CEPEvent("successful cep hit"));}
            }
        );