眨眼 CEP 贪婪匹配

Question

我正在与 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 5 6">

但是，它匹配">

5 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 5 5 6"按预期工作。

是否有可能让贪婪的匹配者在没有不同起始模式的情况下抓住所有比赛？

还是我错过了什么？

斯蒂芬

Answer 1

感谢Chesnay Schepler在上面的评论：

有一个

关于贪婪匹配的已知错误可以解释这种行为：issues.apache.org/jira/browse/FLINK-8914

我暂时会注意到这一点作为答案。

Answer 2

要控制事件将被分配的匹配项数，您需要指定名为 AfterMatchSkipStrategy 的跳过策略。

use Pattern.begin("start"， AfterMatchSkipStrategy.skipPastLastEvent(((

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");