是否可以在 apache flink CEP 中处理多个流?

Question

我的问题是，如果我们有两个原始事件流，即烟雾和温度，并且我们想通过将运算符应用于原始流来找出复杂事件(即火灾(是否发生了，我们可以在 Flink 中做到这一点吗？

我之所以问这个问题，是因为到目前为止，我看到的所有 Flink CEP 示例都只包含一个输入流。如果我错了，请纠正我。

Answer 1

简答- 是的，您可以根据来自不同流源的事件类型读取和处理多个流和触发规则。

长答案- 我有一个有点类似的要求，我的答案是基于您正在阅读来自不同 kafka 主题的不同流的假设。

从在单个源中流式传输不同事件的不同主题中读取：

FlinkKafkaConsumer010<BAMEvent> kafkaSource = new FlinkKafkaConsumer010<>(
Arrays.asList("topicStream1", "topicStream2", "topicStream3"),
new StringSerializerToEvent(),
props);
kafkaSource.assignTimestampsAndWatermarks(new 
TimestampAndWatermarkGenerator());
DataStream<BAMEvent> events = env.addSource(kafkaSource)
.filter(Objects::nonNull);

序列化程序读取数据并将其解析为具有通用格式 - 例如。

@Data
public class BAMEvent {
private String keyid;  //If key based partitioning is needed
private String eventName; // For different types of events
private String eventId;  // Any other field you need
private long timestamp; // For event time based processing 
public String toString(){
return eventName + " " + timestamp + " " + eventId + " " + correlationID;
}
}

在此之后，事情就非常简单了，根据事件名称定义规则并比较事件名称以定义规则(您还可以定义复杂的规则，如下所示(：

Pattern.<BAMEvent>begin("first")
.where(new SimpleCondition<BAMEvent>() {
private static final long serialVersionUID = 1390448281048961616L;
@Override
public boolean filter(BAMEvent event) throws Exception {
return event.getEventName().equals("event1");
}
})
.followedBy("second")
.where(new IterativeCondition<BAMEvent>() {
private static final long serialVersionUID = -9216505110246259082L;
@Override
public boolean filter(BAMEvent secondEvent, Context<BAMEvent> ctx) throws Exception {
if (!secondEvent.getEventName().equals("event2")) {
return false;
}
for (BAMEvent firstEvent : ctx.getEventsForPattern("first")) {
if (secondEvent.getEventId = firstEvent.getEventId()) {
return true;
}
}
return false;
}
})
.within(withinTimeRule);

我希望这能让您想到将一个或多个不同的流集成在一起。

Answer 2

我想知道是否可以进行严格链接(如果可以使用下一个，则不要遵循(，因为在给定的流中，特定时间戳可能有许多事件。所以说时间 t1-： a，b，c - 这三个事件来了，时间 t2-： a2，b2，c2 来到 flink 引擎。所以，我想知道我们如何得到 event(a(.next(a2(，因为可能永远不会是这样，因为系列会像 -： 一个 b c 答2 B2 C2

但是，如果 CEP 模块处理事件，以便将一个时间戳视为单个事件，则这是有意义的。