我想做;从映射结构中包含的任何数组列表元素的 CEP 开始，并继续使用我已启动的其余数组列表元素。映射和图案结构：final Map< Integer,ArrayList<String>> deger = new HashMap<Integer,ArrayList<String>>(); deger.put(1,new ArrayList<String>(Arrays.asList("h:1","l:1","g:0"))); deger.put(2,new ArrayList<String>(Arrays.asList("h:1","l:1","g:1"))); deger.put(3,new ArrayList<String>(Arrays.asList("h:2","l:3","g:1"))); deger.put(4,new ArrayList<String>(Arrays.asList("h:0","l:2","g:2"))); for(int i=1;i<deger.size()+1;i++) { temp1.add(deger.get(i)); }Pattern<String,?> pattern = Pattern.<String>begin("start").where( new SimpleCondition<String>() {// @Override public boolean filter(String value) throws Exception { for (ArrayList<String> aa: temp1){ for (String dd : aa) if(value.equals(dd)){ return true; } } return false; } } ).followedBy("middle").where( new SimpleCondition<String>() { @Override public boolean filter(String value) throws Exception { return value.equals(temp1.get(1)); } } ).followedBy("end").where( new SimpleCondition<String>() { @Override public boolean filter(String value) throws Exception { return value.equals(temp1.get(2)); } } );我的目的是在映射中使用数组列表元素发出警告，但是数组列表元素的顺序并不重要，因为其中的流流。我想继续处理此数组的其余元素，当我从此处的任何数组开始时，我可以返回此数组的信息。例如：Incoming data = "l:1","h:1","g:0"my pattern = "h:1","l:1","g:0" Start -> l:1 findMiddle -> g:0 or h:1 | h:1 findEnd -> g:0 find -> alarm
2 回答
慕妹3242003
TA贡献1824条经验 获得超6个赞
public static Integer temp1;
public static Map<Integer,ArrayList<String>> temp2 = new HashMap<>();
final Map< Integer,ArrayList<String>> deger = new HashMap<>();
deger.put(1,new ArrayList<>(Arrays.asList("h:1","g:1","s:0")));
deger.put(2,new ArrayList<>(Arrays.asList("h:1","g:1","g:0")));
deger.put(3,new ArrayList<>(Arrays.asList("h:1","c:0","g:0")));
deger.put(4,new ArrayList<>(Arrays.asList("h:1","s:1","g:0")));
Pattern<String,?> pattern = Pattern.<String>begin("start").where(
new SimpleCondition<String>() {
@Override
public boolean filter(String value) throws Exception {
flag = false;
for(Map.Entry<Integer, ArrayList<String>> entryStart : deger.entrySet()) {
if(entryStart.getValue().contains(value) && !temp2.containsKey(entryStart.getKey())){
ArrayList<String> newList = new ArrayList<String>();
newList.addAll(entryStart.getValue());
newList.remove(value);
temp2.put(entryStart.getKey(),newList);
flag = true;
}
}
return flag;
}
}
).followedBy("middle").where(
new SimpleCondition<String>() {
@Override
public boolean filter(String middle) throws Exception {
flag = false;
for(Map.Entry<Integer, ArrayList<String>> entryMiddle : temp2.entrySet()) {
if(entryMiddle.getValue().contains(middle) && entryMiddle.getValue().size() == 2){
ArrayList<String> newListMiddle = new ArrayList<String>();
newListMiddle.addAll(entryMiddle.getValue());
newListMiddle.remove(middle);
temp2.put(entryMiddle.getKey(),newListMiddle);
flag = true;
}
}
return flag;
}
}
).followedBy("end").where(
new SimpleCondition<String>() {
@Override
public boolean filter(String end) throws Exception {
flag = false;
for(Map.Entry<Integer, ArrayList<String>> entryEnd : temp2.entrySet()) {
if(entryEnd.getValue().contains(end) && entryEnd.getValue().size() == 1){
flag = true;
temp1 = entryEnd.getKey();
}
}
if (flag)
temp2.remove(temp1);
return flag;
}
}
);
PatternStream<String> patternStream = CEP.pattern(stream_itemset_ham,pattern);
DataStream<String> result = patternStream.select(
new PatternSelectFunction<String, String>() {
@Override
public String select(Map<String, List<String>> map) throws Exception {
ArrayList<String> NewList= new ArrayList<>();
NewList.addAll(deger.get(temp1));
String found = "Found";
for (String list_element : NewList)
found += " " + list_element ;
return found;
}
}
);
result.print();
我从你的问题中了解到，可以提供这种解决方案。
慕少森
TA贡献2019条经验 获得超9个赞
所以目前AFAIK Flink不支持开箱即用的无序模式，所以基本上我看到了两种解决这个问题的方法：
1）您可以创建要搜索的所有可能的模式，并简单地合并所有生成的数据流。
2）正如这篇文章所建议的FlinkCEP：我可以引用一个较早的事件来定义后续匹配吗？您可以尝试使用它来允许您访问已经匹配的先前元素，因此基本上您必须定义与列表中所有可能的元素匹配的模式，然后只需检查最后一个条件，如果所有三个元素都属于同一列表。如果是这样，则找到该模式。IterativeCondition
添加回答
举报
0/150
提交
取消