水印策略不适用于Flink中的Kafka消费者

Question

我是Flink的新手，所以在定义Flink中的水印时，我面临一些问题。

让我们从Kafka消费者开始。使用的反序列化是JSONKeyValueDeserializationSchema，因此没有自定义解析。

val kafkaConsumer: FlinkKafkaConsumer[ObjectNode] = new FlinkKafkaConsumer[ObjectNode](
  kafkaTopic,
  new JSONKeyValueDeserializationSchema(false),
  properties
)

如果将接收器应用于此代码，则其工作正常。问题是需要水印来避免无序事件。这就是我写的策略：

 val watermarkStrategy: WatermarkStrategy[ObjectNode] = WatermarkStrategy
  .forBoundedOutOfOrderness[ObjectNode](Duration.ofSeconds(100))
  .withTimestampAssigner(
    new SerializableTimestampAssigner[ObjectNode] {
      override def extractTimestamp(record: ObjectNode, recordTimestamp: Long): Long = {
        Instant.parse(record.get("value").get("content").get("timestamp").asText()).getEpochSecond
      }
    })

在做了一些研究后，我最终得到了这段代码，但这不起作用。这些是我的问题：

• 在这里使用ObjectNode是最好的选择吗？还有其他选择吗
• 字段timestamp是ISO 8601标准中的一个字符串，因此我必须将其解析为long。这是最好的方法吗？有更好的方法吗
• SerializableTimestAssigner和ForBoundedAutoforDerness是否是正确的使用对象

Answer 1

问题是我的语法分析。

dateFormat.parse(record.get("value").get("content").get("timestamp").asText()).getTime