如何阻止Apache Flink CEP模式？

Question

请帮帮我，我有两个问题:
我从Apache Kafka json-messages读到（然后我有步骤：反序列化到POJO、筛选器、keyBy...）

1. 使用哪个更好：KeyedProcessFunction（带有状态、计时器、if-else逻辑块）还是Flink CEP模式库？

我可以检查KeyedProcessFunction中的输入序列（检查state,if-else blocks,out.collect（...）,state.clear（）...你会理解我的），我还可以使用带有条件和量化器的Flink CEP库。

例如:
我有输入序列:A1,（无事件1分钟）A2,（无事件5分钟）AK3,（无事件1分钟）AK4,（无事件5分钟以上）A5。（在A1和A5之间可能有很多事件）
我想发送输出:A1,A3,A5。
第一个事件，如果下一个事件发生在前一个事件之后少于5分钟，它将不发送到输出；如果下一个事件发生在前一个事件之后超过5分钟，它将发送到输出。
我应该向我的模式添加什么？？？

Pattern<Event, ?> pattern = Pattern.
<Event>begin("start")
.where(new SimpleCondition<Event>(){
 public boolean filter(Event event){
return event.getName().contains("A");
}
}).within(Time.minutes(5));

Answer 1

虽然乍一看，这个特定的示例作为KeyedProcessFunction实现起来似乎相当微不足道，但如果消息可以无序到达，肯定会产生一些复杂性。那么你可能会被愚弄，以为会有很大的差距，而实际上根本没有。

但是，如果您想要一个简单的、开箱即用的现成解决方案，这个特定的示例非常适合会话窗口。

对于CEP，我认为一个可行的解决方案会有这样的味道：您正在寻找一个a（称为A1)的序列，然后紧接着另一个a（称为A2)，其中(A2.timestamp-a1.timestamp)>=5分钟。当找到匹配时，发射A1并使匹配引擎前进，使A2成为新的A1。（方便的是，CEP对输入流进行预排序，因此您不必担心会出现混乱。）