Flink CEP-如果序列没有在一段时间内到达，而不考虑任何其他消息的到达，则发出警报

Question

我编写了一个Flink CEP片段，该片段使用松弛的邻接度(followedby)检查状态模式（由ID键控）。其思想是，如果特定状态在指定时间内未在第一个状态之后到达，则发出警报。

这是有效的，但是如果有nomore消息到该流，则不会触发警报。但只有当带有某种随机状态的消息到达时，这部分才会被触发。

那么，我如何使它触发警报，即使没有消息到达这个流，当带有下一个序列的消息没有带时间到达时？

Pattern<Transaction, Transaction> pattern = Pattern.<Transaction>begin("start")
            .where(new SimpleCondition<Transaction>() {

                private static final long serialVersionUID = 1L;

                @Override
                public boolean filter(Transaction value) throws Exception {
                    return value.getStatus().equals(Status.STATUS_1);
                }
            })
           .followedBy("end")
           .where(new SimpleCondition<Transaction>() {
                @Override
                public boolean filter(Transaction value) throws Exception {
                    return value.getStatus().equals(Status.STATUS_2);
                    return amlveri;
                }
            }).within(Time.seconds(15));

PatternStream<Transaction> patternStream = CEP.pattern(dataStreamSource, pattern);
OutputTag<Alert> timedOutPartialMatchesTag = new OutputTag<Alert>("alert",
            TypeInformation.of(Alert.class)) {};
    SingleOutputStreamOperator<Alert> select = patternStream.flatSelect(timedOutPartialMatchesTag,
            new PatternFlatTimeoutFunction<Transaction, Alert>() {

                @Override
                public void timeout(Map<String, List<Transaction>> values, long arg1, Collector<Alert> arg2)
                        throws Exception {
                    Transaction failedTrans = values.get("start").get(0);
                    arg2.collect(new Alert("status_2 didnt arrive in time, ", failedTrans));
                }
            }, new PatternFlatSelectFunction<Transaction, Alert>() {

                @Override
                public void flatSelect(Map<String, List<Transaction>> arg0, Collector<Alert> arg1)
                        throws Exception {
                       // do not do anything
                }
            });
    select.getSideOutput(timedOutPartialMatchesTag).print();

Answer 1

如果您正在使用事件时间，则invin()方法正在等待水印来触发事件时间计时器。但是如果没有事件到达，水印就不会前进（假设您正在使用BoundedOutoFordernesTimeStampExtractor这样的东西来生成水印）。

如果您需要在没有事件到达的情况下检测时间的流逝，那么就需要使用处理时间。您可以将timecharacteristic设置为处理时间，或者实现一个水印生成器，它使用处理时间计时器来人为地提前水印，尽管缺少事件。