如何解决如何广播CEP模式并在CEP中遍历多个模式?
我正在尝试在CEP中的Patternstream上应用模式列表,我在下面尝试了一些附加代码,我是Flink的新手,我不确定这是愚蠢的方法还是正确的方法,我不确定是否可以使用是否在KeyedbroadCastFunction中的processFunction内部创建数据流
public static final MapStateDescriptor<String,String> patternDescriptor = new MapStateDescriptor<String,String>("CEPPatternList",BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO);
StreamExecutionEnvironment env = env.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<JSONObject> source =
env.fromCollection(Arrays.asList(event,event2,event3));//KafkaSource
DataStream<Tuple2<String,JSONObject>> eventDataStream =
source.map(new JSONObjectToTuple2());
DataStream<Tuple2<String,JSONObject>> eventStream = eventDataStream.keyBy(0);
DataStream<Tuple2<String,String>> patternStream =
env.fromElements(pattern).flatMap(new FlatMapFunction<String,Tuple2<String,String>>() {
@Override
public void flatMap(String value,Collector<Tuple2<String,String>> out) throws Exception {
out.collect(new Tuple2<>("PATTERN_1",value));
}
});
broadcastStream<Tuple2<String,String>> broadcastStream =
patternStream.broadcast(patternDescriptor);
DataStream<Tuple2<String,JSONObject>> output =
eventStream.connect(broadcastStream).process(new KeyedbroadcastProcessFunction<String,JSONObject>,String>,JSONObject>>() {
@Override
public void processElement(Tuple2<String,JSONObject> value,ReadOnlyContext ctx,JSONObject>> out) throws Exception {
for (Map.Entry<String,String> patterns :
ctx.getbroadcastState(patternRuleDescriptor).immutableEntries()) {
String patternValue = patterns.getValue();
DataStream<Tuple2<String,JSONObject>> eventStream =
env.fromElements(value);
PatternStream<Tuple2<String,JSONObject>> patternStream =
cepPatternMatching.compile(patternValue,eventStream);
OutputTag<Tuple2<String,JSONObject>> timedout = new OutputTag<Tuple2<String,JSONObject>>(
"timedout") {
};
SingleOutputStreamOperator<Tuple2<String,JSONObject>> result = patternStream.flatSelect(
timedout,new EventTimeOut(),new PatternFlatSelect()
);
result.flatMap(new FlatMapFunction<Tuple2<String,Object>() {
@Override
public void flatMap(Tuple2<String,Collector<Object> out) throws Exception {
out.collect(value);
}
});
ctx.output(unMatched,value);
}
}
}
@Override
public void processbroadcastElement(Tuple2<String,String> value,Context ctx,JSONObject>> out) throws Exception {
System.out.println("Pattern Name: " + value.f0);
System.out.println("Pattern Condition: " + value.f1);
ctx.getbroadcastState(patternDescriptor).put(value.f0,value.f1);
}
});
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: The implementation of the StreamExecutionEnvironment is not serializable. The object probably contains or references non serializable fields.
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:151)
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:71)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1821)
at org.apache.flink.streaming.api.datastream.broadcastConnectedStream.clean(broadcastConnectedStream.java:249)
at org.apache.flink.streaming.api.datastream.broadcastConnectedStream.process(broadcastConnectedStream.java:162)
at org.apache.flink.streaming.api.datastream.broadcastConnectedStream.process(broadcastConnectedStream.java:139)
at CEPPatternMatchingApp.main(CEPPatternMatchingApp.java:122)
Caused by: java.io.NotSerializableException: org.apache.flink.streaming.api.environment.LocalStreamEnvironment
Caused by: java.io.NotSerializableException: org.apache.flink.streaming.api.environment.LocalStreamEnvironment
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:586)
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:133)
... 7 more
我没有在代码中的任何地方进行序列化,不确定为什么会出现此错误,我什至尝试实现可序列化的接口,但仍然出现错误
能帮我吗,有办法广播CEP模式并逐一迭代以应用于数据流
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。