这是我的代码。
SplitStream<MonitoringEvent> splitStream = inputStream.split(new OutputSelector<MonitoringEvent>() {
@Override
public Iterable<String> select(MonitoringEvent me) {
List<String> ml = new ArrayList<String>();
ml.add(me.getEventType());
return ml;
}
我有随机顺序的监控事件流温度：80，压力：70，湿度：80，温度：30…
使用上面的代码，我按事件类型划分流，即温度流、压力流。
问题是，如果我知道eventType，我可以从类似的splitStream中选择它
splitStream.select('temperatureStream')
但是eventType是动态的并且不是预定义的。
我将如何将CEP应用于此动态流。如果
temperate is > 90 for past 10 minutes ...
pressure is > 90 for past 10 minutes ...
如果我错了，请纠正我，但我认为由于flink的并行性，不可能对select进行动态查找。你的程序会被翻译成flinks任务经理的并行指令，而jobmanager会协调这些操作。如果没有对抽象语法树的全面了解，视差就根本无法应用。。。也许你可以找到一些共同的属性，所有的消息共享和不同的