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

Question

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

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

Answer 1

我想知道是否可以进行严格的链接（如果可以使用next，则不必遵循），因为在给定的流中，对于特定的时间戳可能会有许多事件。假设时间t1-：a，b，c-，这三个事件发生，时间t2-：a2，b2，c2发生在flink引擎上。所以，我想知道我们如何得到事件（a）。下一个（a2），因为它可能永远不会是这样，因为序列类似于-：a b c a2 b2 c2

然而，如果CEP模块处理事件时将一个时间戳视为单个事件，那么这是有意义的。

Answer 2

简短回答-是的，您可以根据不同流源中的事件类型读取和处理多个流和火灾规则。

长答案-我有一个有点类似的要求，我的答案是基于假设你正在阅读来自不同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);

我希望这能为您提供将一个或多个不同流集成在一起的想法。