我想要做;从CEP开始，使用地图结构中包含的任何arraylist元素，并继续使用我已经开始的其他arraylist元素。地图和图案结构：
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));
}
}
);
我的目的是在地图中给出arraylist元素的警告，但arylylist元素的顺序并不重要，因为其中有流。我想继续这个数组的其余元素，我可以返回此数组的信息时我从这里开始使用任何数组。例如：
Incoming data = "l:1","h:1","g:0"
my pattern = "h:1","l:1","g:0"
Start -> l:1 find
Middle -> g:0 or h:1 | h:1 find
End -> g:0 find -> alarm
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();
我从你的问题中了解到，可以提供这种解决方案。
所以目前AFAIK Flink不支持开箱即用的有序模式，所以基本上我看到了解决这个问题的两种方法：
1）您可以创建要搜索的所有可能模式，并简单地将所有结果数据流合并。
2）正如这篇文章建议FlinkCEP: Can I reference an earlier event to define a subsequent match?您可以尝试使用IterativeCondition，它将允许您访问已经匹配的先前元素，所以基本上您必须定义匹配列表中所有可能元素的模式，然后只需检查最后一个条件如果他们三个都属于同一个名单。如果是这样，则找到模式。