Apache Flink模式检测未找到任何匹配项

Question

我正在尝试使用Apache Flink CEP（复杂事件处理）库捕获模式。我从以下结构开始，我希望看到id[1,2]和[3,4]的2个匹配项。但是我没有看到任何结果。

public class StreamingJob {

    private static Logger logger = LoggerFactory.getLogger(StreamingJob.class);

    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();

        ArrayList <Event> strings = new ArrayList <>();
        strings.add(new Event(0L, "room", 9));
        strings.add(new Event(1L, "room", 10));
        strings.add(new Event(2L, "garden", 11));
        strings.add(new Event(3L, "room", 12));
        strings.add(new Event(4L, "garden", 13));
        strings.add(new Event(5L, "room", 14));
        strings.add(new Event(6L, "room", 15));


        KeyedStream <Event, String> source = env.fromCollection(strings).keyBy(Event::getName);

        source.print("###-source");

        Pattern <Event, ?> pattern = Pattern. <Event>begin("room")
                .where(new SimpleCondition <Event>() {
                    @Override
                    public boolean filter(Event value) {
                        logger.info("### value: {}", value);
                        return value.getName().equals("room");
                    }
                })
                .next("garden")
                .where(new SimpleCondition <Event>() {
                    @Override
                    public boolean filter(Event value) {
                        logger.info("### value: {}", value);
                        return value.getName().equals("garden");
                    }
                });

        PatternStream <Event> patternStream = CEP.pattern(source, pattern);


        // process 
        DataStream <Alarm> result = patternStream.process(
                new PatternProcessFunction <Event, Alarm>() {
                    @Override
                    public void processMatch(
                            Map <String, List <Event>> pattern,
                            Context ctx,
                            Collector <Alarm> out) throws Exception {
                        logger.info("### pattern: {}", pattern);
                        logger.info("### ctx: {}", ctx);
                        out.collect(new Alarm(pattern.toString()));
                    }
                });

        result.print("###");

        // or select function
        patternStream.select(new PatternSelectFunction <Event, Alarm>() {
            @Override
            public Alarm select(Map <String, List <Event>> pattern) throws Exception {
                logger.info("###");
                return new Alarm(pattern.toString());
            }
        }).print("###");

        // execute program
        env.execute("Flink Streaming Java API Skeleton");
    }
}

source.print（）方法正在打印源流，对于接收器，我尝试了进程和选择方法，但都无法打印结果。此外，我的过滤器或匹配方法的日志根本没有打印出来。我的印象是过滤器功能甚至没有使用。Event和警报对象是简单的pojos，如下所示：

public class Event implements Serializable {
    Long id;
    String name;
    Integer temperature;
    //...
}

public class Alarm implements Serializable {
    String text;
    // ...
}

我还尝试将其更改为过滤其他字段，例如，使用温度字段，我想捕获奇数-偶数数字序列，但仍然没有打印任何结果。

Answer 1

CEP依赖于能够按时间戳对事件流进行排序。这要求您要么提供Watermark Strategy（如果您想使用事件时间戳），要么使用流转时长语义学指定您希望模式匹配按摄取顺序完成。

您可以通过进行以下小改动来实现后者：

PatternStream <Event> patternStream = 
  CEP.pattern(source, pattern).inProcessingTime();