Flink CEP事件未触发

Question

我已经在Flink中实现了CEP模式，它按预期工作连接到本地Kafka代理。但是当我连接到基于集群的云kafka设置时，Flink CEP不会触发。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //saves checkpoint
    env.getCheckpointConfig().enableExternalizedCheckpoints(
            CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

我正在使用AscendingTimestampExtractor，

consumer.assignTimestampsAndWatermarks(
    new AscendingTimestampExtractor<ObjectNode>() {
      @Override
      public long extractAscendingTimestamp(ObjectNode objectNode) {
        long timestamp;
        Instant instant = Instant.parse(objectNode.get("value").get("timestamp").asText());
        timestamp = instant.toEpochMilli();
        return timestamp;
      }
    });

我也收到警告消息，

AscendingTimestampExtractor：140-违反时间戳单调性：1594017872227

而且我也尝试过使用Assignerwith周期水印和AssignerSusPunctuatedWatermark没有一个是工作

我附上了Flink控制台屏幕截图，其中水印未指定。更新的flink控制台屏幕截图

有人能帮忙吗？

Answer 1

CEP必须首先根据水印对输入流进行排序。所以问题可能出在水印上，但您还没有向我们展示足够的信息来调试原因。一个常见的问题是有一个空闲源，这会阻止水印前进。

但还有其他可能的原因。为了调试这种情况，我建议您查看一些指标，无论是在Flink Web UI中，还是在连接了指标系统的情况下在指标系统中。首先，通过查看管道不同阶段的numRecordsIn、numRecordsOut或numRecordsInPerSecond和numRecordsOutPerSecond检查记录是否在流动。

如果有事件，请在作业的不同任务中查看currentOutputWatermark，以查看事件时间是否提前。

更新时间：

您可能正在Kafka使用者上调用assignTimestampsAndWatermarks，这将导致每个分区的水印。在这种情况下，如果您有一个空闲分区，该分区将不会产生任何水印，这将保留整个水印。尝试在源代码生成的数据流上调用assignTimestampsAndWatermarks，看看这是否解决了问题。（当然，如果没有逐分区水印，您将无法使用AscendingTimestampExtractor，因为流将不有序。）