使用自定义触发器与Flink CEP模式API的全局窗口中的全局窗口

Question

我是Apache Flink API的新手，我正在尝试理解它所提供的不同窗口。

我有一系列事件，例如：

    device_id,trigger_id,event_time,messageId
    1,START,1520433909396,1
    1,TRACKING,1520433914398,2
    1,TRACKING,1520433919398,3
    1,STOP,1520433924398,4
    1,START,1520433929398,5
    1,TRACKING,1520433934399,6
    1,TRACKING,1520433939399,7
    1,TRACKING,1520433944399,8
    1,STOP,1520433949399,9

trigger_id可以是一个指示器，例如： start ，跟踪， stop

我想做的是基于device_id组所有传入事件，并根据trigger_id定义窗口。即从开始到停止的所有事件，然后进行一些计算，例如：平均，最大等。

这可以定义为GlobalWindow和基于Trigger_ID的自定义触发器，并使用自定义驱动器每次驱逐事件列表停止触发器已达到。

另一个选择可能是使用Flink CEP。我定义了以下模式

DataStream<String> input = env.readTextFile("events.csv");
    // create event stream
    DataStream<Event> events = input.map(new LineToEvent());
    DataStream<Event> waterMarkedStreams = events.assignTimestampsAndWatermarks(new EventAssigner());
    Pattern<Event, Event> tripPattern =
            Pattern.<Event>begin("start",  AfterMatchSkipStrategy.noSkip())
                    .where(START_CONDITION)
                    .followedBy("middle").where(MIDDLE_CONDITION).oneOrMore()
                    .followedBy("end").where(END_CONDITION);
    PatternStream<Event> patternStream = CEP.pattern(waterMarkedStreams, tripPattern);
    DataStream<String> result = patternStream.select(
            new PatternSelectFunction<Event, String>() {
                @Override
                public String select(Map<String, List<Event>> pattern) throws Exception {
                    StringBuilder builder = new StringBuilder();
                    builder.append(pattern.get("start").get(0).getMessageId()).append(",");
                    List<Event> vals = pattern.get("middle");
                    for (Event e: vals) {
                        builder .append(e.getMessageId()).append(",");
                    }
                    builder.append(pattern.get("end").get(0).getMessageId()).append(",");
                    return builder.toString();
                }
            });
    result.print();

所有条件都是静态内部类实现simperecondition

但是，该模式与事件流上的Alll可能的解决方案匹配，因此...

    1> 1,2,3,4,
    1> 1,2,3,6,9,
    2> 1,2,4,
    2> 5,6,7,8,9,
    3> 1,2,3,6,7,8,9,
    3> 5,6,7,9,
    4> 1,2,3,6,7,9,
    4> 5,6,9,

模式有驱逐的概念吗？如何仅保留特定事件集。即。

1,2,3,4,
5,6,7,8,9,

Answer 1

是的，您应该能够使用CEP进行此操作。您需要做的不仅仅是定义模式，还必须将图案应用于流，然后选择匹配序列并使用这些序列来发出一些结果。文档中有一个相当完整的示例。