Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
.next("middle").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("error");
}
}).followedBy("end").where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event value) throws Exception {
return value.getName().equals("critical");
}
}).within(Time.seconds(10));
有没有办法用我通过Event
传入的value.getSomeTimeField()
替换Time.seconds(10)
?
我想你想以事件时间的方式工作。有关它的更多信息,您可以查看此文档和有关如何从元素中提取时间戳的本节。
在您的示例中,您可以执行以下操作:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Event> input = ...
input.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Event>() {
@Override
public long extractAscendingTimestamp(MyEvent element) {
return value.getSomeTimeField();
}
})
CEP.pattern(input, pattern).select(...)
这样,事件将在流中自动排序,并且超时将在两种情况下应用与时间字段相关的时间字段。