我想在一个数据流中连接两个Kafka主题。
事实上,两个数据流必须具有相同的id才能进行连接。事件是来自传感器的数据,规则包含将使用cep(来自用户界面)检查的规则。
这是我的考试,但不起作用,有人能帮我吗?
DataStream<Object> evtAndRule=inputEventStream.join(rulesStream)
.where(new KeySelector<TrackEvent, Object>() {
@Override
public Object getKey(Event event) throws Exception {
return event.getId();
}
}).equalTo(new KeySelector<RulesEvent, Object>() {
@Override
public Object getKey(RulesEvent rulesEvent) throws Exception {
return rulesEvent.getId();
}
}).window(TumblingTimeWindows.of(Time.of(10, TimeUnit.SECONDS)))
.apply(new FlatJoinFunction<TrackEvent, RulesEvent, Object>() {
@Override
public void join(TrackEvent trackEvent, RulesEvent rulesEvent, Collector<Object> collector) throws Exception {
....
}
});
1条答案
按热度按时间dxxyhpgq1#
我试过这个,但我不知道如何检索所需的规则,如果这是最好的解决方案