如何使用flink SQL识别是否已到达N个特定事件？

问题描述

我对flink感到陌生，并试图达到以下要求。

我的要求： 我的应用程序正在监听来自事件中心的事件实时流。我需要做->当事件EVT1和EVT2都到达时（以任何顺序），然​​后根据hashmap（字符串，列表）中的键将它们分组。现在，EVT1和EVT2可以从事件中心多次出现。我需要使用Flink sql方式实现。

我到目前为止所做的事情： 我的sql是：

select * from Events
Where name in ('EVT1','EVT2')
and (select count(*) from Events where name = 'EVT1') > 0
and (select count(*) from Events where name = 'EVT2') > 0;

我的Java代码段是-

Table table = env.sqlQuery(sql);
env.toRetractStream(table,Row.class)
 .process(new ProcessFunction <Tuple2<Boolean,Row>,Object>() {
     @Override
     public void processElement(Tuple2<Boolean,Row> value,Context ctx,Collector<Object> out) throws Exception {
         Row ev = value.f1;
         log.info(ev);
         // more code here
     }    
 });

我的测试场景：-

1. 我启动Flink程序
从活动中心收到
2. EVT1。什么都没发生。
从活动中心收到
3. EVT2。条件匹配，log.info（）依次打印2条记录。
4. 1小时后，事件EVT1再次到达。条件仍然匹配并在日志中打印此新的EVT1。

我的问题：-

1. 在上面的步骤4中，为什么条件匹配。是因为 新的EVT1到来，flink在旧的EVT2上将其匹配 较早处理？如果是这样，我该如何防止呢？步骤4之后，我要 flink等待下一个匹配的EVT2。而且，即使它匹配 旧事件，那么为什么只打印新的EVT1。
2. 可以使用Flink解决此要求吗？怎么样？
3. flink可以一口气为我提供所有结果（如清单所示）。因为在上面的示例中 在processElement（）方法中，我怎么知道我已经处理了所有 记录。因为在实际的生产场景中，我不会知道 我的sql要匹配的许多事件。

谢谢您的帮助。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）