未遵循的 Apache Flink CEP 模式操作答案

【问题标题】：Apache Flink CEP Pattern operation for NOT followedBy未遵循的 Apache Flink CEP 模式操作
【发布时间】：2017-07-25 20:31:15
【问题描述】：

我有一个场景，如果第二个事件在 x 秒内没有跟随第一个事件，我必须更改状态。例如用户在 100 分钟内没有注销，认为他处于无效状态。如何使用当前的模式操作来设计？

【问题讨论】：

标签： apache-flink flink-cep

【解决方案1】：

由于这已经实施，我想为那些来这里寻找答案的人回答这个问题。

从 Flink 1.0.0 开始，这可以通过处理超时模式来完成，例如，如果您的 CEP 模式是这样的：

部分来自Flink Website 的示例（1.2 和 1.3 之间有一些重大变化，请相应地调整您的代码，此答案侧重于 1.3）

模式描述： - 获取第一个“错误”类型的事件，然后在 10 秒内获取第二个“关键”类型的事件

Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
.next("middle").where(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) throws Exception {
        return value.getName().equals("error");
    }
}).followedBy("end").where(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) throws Exception {
        return value.getName().equals("critical");
    }
}).within(Time.seconds(10));

PatternStream<BAMEvent> patternStream = CEP.pattern(inputStream, pattern)

DataStream<Either<String, String>> result = patternStream.select(new PatternTimeoutFunction<Event, String>() {
  @Override
  public String timeout(Map<String, List<Event>> map, long l) throws Exception {
    return map.toString() +" @ "+ l;
  }
}, new PatternSelectFunction<Event, String>() {

  @Override
  public String select(Map<String, List<Event>> map) throws Exception {
    return map.toString();
  }
});

对于这种情况，如果用户在 100 分钟后仍未注销，则由于相应的事件不会到达，这将导致模式超时，部分事件（启动事件）将被捕获PatternTimeoutFunction。

【讨论】：

【解决方案2】：

目前这是不可能的。解决方案是有一个超时处理程序，只要事件序列因为超出定义的时间窗口而被丢弃，就会触发该处理程序。已经有一个跟踪 timeout handler 实现的 JIRA 问题。

【讨论】：