我正在和 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");
}
});
但是,(因此提供不同的起始匹配)Greedy 运算符通过发出“3 5 5 5 5 6”按预期工作。
是否有可能让一个贪婪的匹配器在没有不同起始模式的情况下抓取所有匹配项?
还是我错过了什么?
斯蒂芬