并行处理Flink CEP中的多种模式

Question

我有以下情况

有2个虚拟机正在向Kafka发送流，CEP引擎正在接收这些流，当单个流满足特定条件时，会生成警告。

目前，CEP正在检查两条流上的相同情况（当心率

 // detecting pattern
        Pattern<joinEvent, ? > pattern = Pattern.<joinEvent>begin("start")
                .subtype(joinEvent.class).where(new FilterFunction<joinEvent>() {
                    @Override
                    public boolean filter(joinEvent joinEvent) throws Exception {
                        return joinEvent.getHeartRate() > 65 ;
                    }
                })
                .subtype(joinEvent.class)
                .where(new FilterFunction<joinEvent>() {
                    @Override
                    public boolean filter(joinEvent joinEvent) throws Exception {
                        return joinEvent.getRespirationRate() > 68;
                    }
                }).within(Time.milliseconds(100));

但我想对这两个流使用不同的条件。例如，如果

For patient A : if heart rate > 65 and Respiration Rate > 68
For patient B : if heart rate > 75 and Respiration Rate > 78

如何实现这一点？我需要在同一环境中创建多个流环境或多个模式吗？

Answer 1

根据您的需求，如果需要，您可以创建两种不同的模式来进行清晰的分离。

如果您想以相同的模式执行此操作，那么它也是可能的。为此，请在一个kafka源中阅读所有kafka主题：

    FlinkKafkaConsumer010<JoinEvent> kafkaSource = new FlinkKafkaConsumer010<>(
        Arrays.asList("topic1", "topic2"),
        new StringSerializerToEvent(),
        props);

在这里，我假设两个主题的事件结构是相同的，并且您有患者姓名以及事件的一部分。

一旦这样做了，就变得很容易了，因为您只需要创建一个带有“或”的模式，如下所示：

    Pattern.<JoinEvent>begin("first")
        .where(new SimpleCondition<JoinEvent>() {

          @Override
          public boolean filter(JoinEvent event) throws Exception {
            return event.getPatientName().equals("A") && event.getHeartRate() > 65 && joinEvent.getRespirationRate() > 68;
          }
        })
        .or(new SimpleCondition<JoinEvent>() {

          @Override
          public boolean filter(JoinEvent event) throws Exception {
            return event.getPatientName().equals("B") && event.getHeartRate() > 75 && joinEvent.getRespirationRate() > 78;
          }
        });

这将在您的条件匹配时产生匹配。尽管如此，我并不确定在您的示例中“.within（Time.millises（100））”实现了什么。