flink cep贪婪匹配

7z5jn7bk  于 2021-06-25  发布在  Flink
关注(0)|答案(2)|浏览(342)

我正在和flink cep贪婪的操作员打一场仗。
给出以下java代码:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    List<String> strings = Arrays.asList("1,3,5,5,5,5,6,".split(","));

    DataStream<String> input = env.fromCollection(strings);

    Pattern<String, ?> pattern = Pattern.<String>
    begin("start").where(new SimpleCondition<String>() {
        @Override
        public boolean filter(String value) throws Exception {
            return value.equals("5");
        }
    }).oneOrMore().greedy()
    .followedBy("end").where(new SimpleCondition<String>() {

        @Override
        public boolean filter(String value) throws Exception {
            return value.equals("6");
        }
    });

    PatternStream<String> patternStream = CEP.pattern(input, pattern);

    DataStream<String> result = patternStream.select(new PatternSelectFunction<String, String>() {
        @Override
        public String select(Map<String, List<String>> pattern) throws Exception {
            System.err.println("=======");
            pattern.values().forEach(match -> match.forEach(event -> System.err.println(event)));
            System.err.println("=======");
            return "-";
        }
    });

    result.print();
    env.execute("Flink Streaming Java API Skeleton");

我想看看:只有“5 5 5 6”发射
但是,它匹配“5 5 5 6”、“5 5 5 6”、“5 5 6”、“5 6”
如果我这样做了:

begin("start").where(new SimpleCondition<String>() {
        @Override
        public boolean filter(String value) throws Exception {
            return value.equals("3");
        }
    }).followedBy("middle").where(new SimpleCondition<String>() {
        @Override
        public boolean filter(String value) throws Exception {
            return value.equals("5");
        }
    }).oneOrMore().greedy()
    .followedBy("end").where(new SimpleCondition<String>() {

        @Override
        public boolean filter(String value) throws Exception {
            return value.equals("6");
        }
    });

然而,(因此提供了一个不同的起始匹配)贪婪操作符通过发出“3 5 5 6”来按预期工作。
有没有可能让一个贪婪的匹配者在没有不同启动模式的情况下抓取所有的匹配?
还是我遗漏了什么?
斯蒂芬

4nkexdtk

4nkexdtk1#

感谢chesnay schepler的上述评论:
贪婪匹配有一个已知的bug可以解释这种行为:issues.apache.org/jira/browse/flink-8914
我将把这个作为暂时的答案。

83qze16e

83qze16e2#

要控制一个事件将分配多少个匹配项,需要指定名为aftermatchskipstrategy的跳过策略。
使用pattern.begin(“start”,aftermatchskipstrategy.skipcastlastEvent())

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

List<String> strings = Arrays.asList("1,3,5,5,5,5,6,".split(","));

DataStream<String> input = env.fromCollection(strings);

Pattern<String, ?> pattern = Pattern.<String>
        begin("start", AfterMatchSkipStrategy.skipPastLastEvent()).where(new SimpleCondition<String>() {
  @Override
  public boolean filter(String value) throws Exception {
    return value.equals("5");
  }
}).oneOrMore().greedy()
        .followedBy("end").where(new SimpleCondition<String>() {
          @Override
          public boolean filter(String value) throws Exception {
            return value.equals("6");
          }
        });

PatternStream<String> patternStream = CEP.pattern(input, pattern);

DataStream<String> result = patternStream.select(new PatternSelectFunction<String, String>() {
  @Override
  public String select(Map<String, List<String>> pattern) throws Exception {
    System.err.println("=======");
    pattern.values().forEach(match -> match.forEach(event -> System.err.println(event)));
    System.err.println("=======");
    return "-";
  }
});

result.print();
env.execute("Flink Streaming Java API Skeleton");

相关问题