Flink WarterMark 源码信息 未完结

Posted 2022-11-25 青冬

序

参考：

Flink源码分析——WaterMark源码分析 - 知乎

Flink源码解析系列-- WatermarkGenerator接口及其常用实现_打酱油的葫芦娃的博客-CSDN博客

Flink教程-聊聊 flink 1.11 中新的水印策略 - 知乎

EventTime

我们一般使用 EventTime 来进行各种数据的统计，但是数据从产生到进入 Flink 的过程中，经常遇到各种错序问题，那么就需要使用 WaterMark 来进行解决，我们先看看最简单的一个根据 EventTime 开窗进行统计的示例程序：

在这个示例里面，我们构建了 <某用户ID，时间戳> 的一个数据流，然后自定义了 AutoWatermarkInterval、assginTimestampsAndWatermarks、TimestampAssigner，然后根据用户ID进行开窗，进行每4秒钟统计一次 count。

WatermarkStrategy

在 assignTimestampsAndWatermarks 中，需要传入一个 watermarkStrategy，那么我们先来看看这个 watermarkStrategy 是干什么用的。

根据备注我们可以看到，这个接口主要是在 sources 生成 watermark，并且怎么进行记录内部的时间戳的（通过继承 TimestampAssignerSupplier 得到）。

这里面有一些静态方法，和一些 default 方法，提供了一些默认的实现规则供我们调用。

除了 createWatermarkGenerator。

createWatermarkGenerator

提供了一个需要实现的接口方法，来装填一个 WatermarkGenerator。使用这个 WatermarkGenerator 来创建时间戳。

createTimestampAssigner

提供一个 RecordTimestampAssigner 来进行数据的注册 todo。

withTimestampAssigner

传入一个 TimestampAssignerSupplier，或者 SerializableTimestampAssigner 来注册WatermarkStrategyWithTimestampAssigner 。

withIdleness

按照注释说明，这个主要是可以进行空闲的 Source 检测，如果超时没有数据接入，那么就会标记为 idle 状态，不会造成下游数据的堵塞。

使用的 WatermarkStrategyWithIdleness 进行创建，createWatermarkGenerator 使用的是 WatermarksWithIdleness 进行创建。

org.apache.flink.api.common.eventtime.WatermarkStrategyWithIdleness

org.apache.flink.api.common.eventtime.WatermarksWithIdleness

WatermarksWithIdleness 使用 IdlenessTimer 来进行计时操作，如果超时后，在 onPeriodicEmit(output) 中，将 output 设置为 Idle()状态。

assignTimestampsAndWatermarks

org.apache.flink.streaming.api.datastream.DataStream#assignTimestampsAndWatermarks(org.apache.flink.api.common.eventtime.WatermarkStrategy<T>)

获取到数据源后，我们进行了 Watermark 的计算方式就是使用的这个进行的。

还是一样看源码：

通过传入的 watermarkStratey，获取当前的并行度，构建一个 TimestampsAndWatermarksTransformation。 然后返回链式调用。

Watermark

org.apache.flink.api.common.eventtime.Watermark

看看源码：