如果在apache flink中5秒内发现5次相同的数据,则生成cep事件

ifmq2ha2  于 2021-06-21  发布在  Flink
关注(0)|答案(1)|浏览(352)

如果我们发现字符串消息在5秒内连续5次以字符“a”开头,我需要生成cep事件。
为此,我编写了一个类cepchareventpublisher.java,它将把字符串消息(如下发布的消息)发布到kafka主题“charevent”
已发布消息:

b; date- 2019-06-27 09:05:09.605
a; date- 2019-06-27 09:05:10.160
c; date- 2019-06-27 09:05:10.661
b; date- 2019-06-27 09:05:11.162
c; date- 2019-06-27 09:05:11.669
b; date- 2019-06-27 09:05:12.175
b; date- 2019-06-27 09:05:12.675
b; date- 2019-06-27 09:05:13.176
a; date- 2019-06-27 09:05:13.676
c; date- 2019-06-27 09:05:14.176
b; date- 2019-06-27 09:05:14.677
b; date- 2019-06-27 09:05:15.177
b; date- 2019-06-27 09:05:15.678
c; date- 2019-06-27 09:05:16.178
a; date- 2019-06-27 09:05:16.679
c; date- 2019-06-27 09:05:17.179
c; date- 2019-06-27 09:05:17.680
c; date- 2019-06-27 09:05:18.180
c; date- 2019-06-27 09:05:18.681
c; date- 2019-06-27 09:05:19.181
c; date- 2019-06-27 09:05:19.681
a; date- 2019-06-27 09:05:20.182
c; date- 2019-06-27 09:05:20.682
b; date- 2019-06-27 09:05:21.182
c; date- 2019-06-27 09:05:21.682
b; date- 2019-06-27 09:05:22.183
a; date- 2019-06-27 09:05:22.683
b; date- 2019-06-27 09:05:23.184
a; date- 2019-06-27 09:05:23.684
c; date- 2019-06-27 09:05:24.184
b; date- 2019-06-27 09:05:24.685
b; date- 2019-06-27 09:05:25.186
c; date- 2019-06-27 09:05:25.687
b; date- 2019-06-27 09:05:26.187
a; date- 2019-06-27 09:05:26.687
a; date- 2019-06-27 09:05:27.188
a; date- 2019-06-27 09:05:27.688
b; date- 2019-06-27 09:05:28.188
b; date- 2019-06-27 09:05:28.688

现在我有了一个consumer cepchareventconsumer.java,它将读取来自kafka主题charevent的消息,并过滤以字符“a”开头的消息。
然后,我编写了以下模式来生成cep事件/警报,同时我们发现连续5条消息,该消息在5秒内以字符“a”开头。

Pattern<String, String> pattern = Pattern.<String> begin("start")
                .times(5).greedy().where(new SimpleCondition<String>() {
                    private static final long serialVersionUID = -6301755149429716724L;

                    @Override
                    public boolean filter(String value) throws Exception {
                        return value.split(";")[0].equals("a");
                    }
                }).within(Time.seconds(5));

打印以下cepchareventconsumer.java接收的以字符“a”开头的消息。

2> a; date- 2019-06-27 09:05:10.160
1> a; date- 2019-06-27 09:05:13.676
3> a; date- 2019-06-27 09:05:16.679
2> a; date- 2019-06-27 09:05:20.182
3> a; date- 2019-06-27 09:05:22.683
1> a; date- 2019-06-27 09:05:23.684
3> a; date- 2019-06-27 09:05:26.687
1> a; date- 2019-06-27 09:05:27.188
1> a; date- 2019-06-27 09:05:27.688
1> a; date- 2019-06-27 09:05:29.198
2> a; date- 2019-06-27 09:05:30.199

1> a; date- 2019-06-27 09:05:33.703
1> a; date- 2019-06-27 09:05:35.203
3> a; date- 2019-06-27 09:05:36.705
2> a; date- 2019-06-27 09:05:38.207
1> a; date- 2019-06-27 09:05:39.709
2> a; date- 2019-06-27 09:05:40.209
3> a; date- 2019-06-27 09:05:40.728

打印的警报消息:

4> Found: a; date- 2019-06-27 09:05:26.687

在上面的消息“发现:a;日期-2019-06-27 09:05:26.687”是警报消息。
我不明白Flink是怎么在5秒内计算出连续的5条信息的。我觉得有点不对劲。
我附上源代码的git url(flink cep char事件)。谁能按我的要求把它改一下吗。

6rvt4ljy

6rvt4ljy1#

基于cep的应用程序似乎正确地报告了以下5条消息

3> a; date- 2019-06-27 09:05:26.687
1> a; date- 2019-06-27 09:05:27.188
1> a; date- 2019-06-27 09:05:27.688
1> a; date- 2019-06-27 09:05:29.198
2> a; date- 2019-06-27 09:05:30.199

在5秒内发生。
这个 processMatch 方法 PatternProcessFunction 通过了 Map<String, List<String>> match . 对你来说 match.get("start") 返回模式(即整个模式)的“start”子句中5个匹配事件的列表。
因此,要生成一个报告,给出最后一个匹配事件的时间,而不是第一个匹配事件的时间,就需要更改

String start = match.get("start").get(0);
out.collect("Found: " + start);

String last = match.get("start").get(4);
out.collect("Found: " + last);

相关问题