flink的Kafka水印策略在我的应用程序中不起作用

Question

我使用flink版本1.13.0

当我试图使用flink doc的Kafka水印策略时，这似乎不起作用，窗口处理功能将不会运行。

我想知道，在Kafka中，水印的时间戳将使用消费时间还是生产时间？

我的消费者代码如下：

val source = new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), properties)
  .setCommitOffsetsOnCheckpoints(true)
  .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(5)))
stream = env.addSource(source)
  .name(nodeCfg.nodeName)
  .uid(nodeCfg.nodeName)
  .setParallelism(nodeCfg.workerCount)

并像这样使用窗口：

processStream
  .keyBy(_.num)
  .window(TumblingEventTimeWindows.of(Time.seconds(5)))
  .aggregate(new LatTimeAggregate(), new SignLatCalculateProcess())
  .name(nodeCfg.nodeName)
  .uid(nodeCfg.nodeName)
  .setParallelism(nodeCfg.workerCount)
  .addSink(new SignLatSink(serverConfig.smsRuleRedis))
  .name("lat_count_sink")
  .uid("lat_count_sink")

拓扑图是这样的：

Answer 1

由于您没有在水印策略中指定时间戳分配器，因此您依赖FlinkKafkaConsumer为流记录分配时间戳。只有当从Kafka读取的记录的标题中有时间戳时，这才有效。否则，您将需要实现一个时间戳赋值器来从事件中提取时间戳。

请注意，除非您还实现了一个反序列化器，FlinkKafkaConsumer可以使用该反序列化器生成具有时间戳的对象，然后可以提取这些时间戳，否则您将无法实现FlinkKafkaConsumer可以应用的时间戳赋值器。否则，您可以选择在源之后的某个位置应用水印策略。

如果没有时间戳不是问题，那么还有其他问题。例如，您可能有一个空闲的Kafka分区，或者缺少足够远的事件来关闭窗口。

顺便说一下，如果您的事件在每个分区的基础上是有序的，并且如果您在FlinkKafkaConsumer上调用AssignTimestamps和Watermarks（您当前正在这样做），那么您可以使用FormononousTimestamps，而不是ForBoundedAutoforderness，这有一些显著的优点。