Flink水印

Question

在Flink中，我发现了2种设置水印的方法，

第一个是

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.getConfig.setAutoWatermarkInterval(5000)

第二个是

env.addSource(
    new FlinkKafkaConsumer[...](...)
).assignTimestampsAndWatermarks(
   WatermarkStrategy.forBoundedOutOfOrderness[...](Duration.ofSeconds(10)).withTimestampAssigner(...)
)

我想知道哪个最终会生效。

Answer 1

这两者之间根本没有冲突——他们正在处理不同的问题。指定的一切都将生效。

第一个，

env.getConfig.setAutoWatermarkInterval(5000)

指定生成水印的频率（每5000毫秒一个水印）。如果未指定，则将使用默认值200毫秒。

第二个，

env.addSource(
    new FlinkKafkaConsumer[...](...)
).assignTimestampsAndWatermarks(
   WatermarkStrategy.forBoundedOutOfOrderness[...](Duration.ofSeconds(10)).withTimestampAssigner(...)
)

指定如何计算这些水印的详细信息。一、 例如，它们应由FlinkKafkaConsumer使用一种有界的Autoforderness策略生成，有界延迟为10秒。水印策略也需要时间戳赋值器。

没有默认的Watermark Strategy，因此如果您想使用事件时间，则需要类似于第二个代码片段的东西。