概要 #
了解watermark前先需要明确flink支持的时间语义以及为什么需要watermark。
Flink 明确支持以下三种时间语义:
-
事件时间(event time)： 事件产生的时间，记录的是设备生产(或者存储)事件的时间
-
摄取时间(ingestion time)： Flink 读取事件时记录的时间
-
处理时间(processing time)： Flink pipeline 中具体算子处理事件的时间
为了获得可重现的结果，例如在计算过去的特定一天里第一个小时股票的最高价格时，我们应该使用事件时间。这样的话，无论什么时间去计算都不会影响输出结果。然而如果使用处理时间的话，实时应用程序的结果是由程序运行的时间所决定。多次运行基于处理时间的实时程序，可能得到的结果都不相同，也可能会导致再次分析历史数据或者测试新代码变得异常困难。
使用 Event Time #
如果想要使用事件时间，需要额外给 Flink 提供一个时间戳提取器和 Watermark 生成器，Flink 将使用它们来跟踪事件时间的进度。
Watermarks #
让我们通过一个简单的示例来演示为什么需要 watermarks 及其工作方式。
在此示例中，我们将看到带有混乱时间戳的事件流，如下所示。显示的数字表达的是这些事件实际发生时间的时间戳。到达的第一个事件发生在时间 4，随后发生的事件发生在更早的时间 2，依此类推：
··· 23 19 22 24 21 14 17 13 12 15 9 11 7 2 4 →
假设我们要对数据流排序，我们想要达到的目的是：应用程序应该在数据流里的事件到达时就有一个算子（我们暂且称之为排序）开始处理事件，这个算子所输出的流是按照时间戳排序好的。
让我们重新审视这些数据:
(1) 我们的排序器看到的第一个事件的时间戳是 4，但是我们不能立即将其作为已排序的流释放。因为我们并不能确定它是有序的，并且较早的事件有可能并未到达。事实上，如果站在上帝视角，我们知道，必须要等到时间戳为 2 的元素到来时，排序器才可以有事件输出。
需要一些缓冲，需要一些时间，但这都是值得的
(2) 接下来的这一步，如果我们选择的是固执的等待，我们永远不会有结果。首先，我们看到了时间戳为 4 的事件，然后看到了时间戳为 2 的事件。可是，时间戳小于 2 的事件接下来会不会到来呢？可能会，也可能不会。再次站在上帝视角，我们知道，我们永远不会看到时间戳 1。
最终，我们必须勇于承担责任，并发出指令，把带有时间戳 2 的事件作为已排序的事件流的开始
(3)然后，我们需要一种策略，该策略定义：对于任何给定时间戳的事件，Flink 何时停止等待较早事件的到来。
这正是 watermarks 的作用 — 它们定义何时停止等待较早的事件。
Flink 中事件时间的处理取决于 watermark 生成器，后者将带有时间戳的特殊元素插入流中形成 watermarks。事件时间 t 的 watermark 代表 t 之前（很可能）都已经到达。
当 watermark 以 2 或更大的时间戳到达时，事件流的排序器应停止等待，并输出 2 作为已经排序好的流。
(4) 我们可能会思考，如何决定 watermarks 的不同生成策略
每个事件都会延迟一段时间后到达，然而这些延迟有所不同，有些事件可能比其他事件延迟得更多。一种简单的方法是假定这些延迟受某个最大延迟的限制。Flink 将此策略称为 最大无序边界 (bounded-out-of-orderness) watermark。当然，我们可以想像出更好的生成 watermark 的方法，但是对于大多数应用而言，固定延迟策略已经足够了。
延迟 VS 正确性 #
watermarks 给了开发者流处理的一种选择，它们使开发人员在开发应用程序时可以控制延迟和完整性之间的权衡。与批处理不同，批处理中的奢侈之处在于可以在产生任何结果之前完全了解输入，而使用流式传输，我们不被允许等待所有的时间都产生了，才输出排序好的数据，这与流相违背。
我们可以把 watermarks 的边界时间配置的相对较短，从而冒着在输入了解不完全的情况下产生结果的风险-即可能会很快产生错误结果。或者，你可以等待更长的时间，并利用对输入流的更全面的了解来产生结果。
当然也可以实施混合解决方案，先快速产生初步结果，然后在处理其他（最新）数据时向这些结果提供更新。对于有一些对延迟的容忍程度很低，但是又对结果有很严格的要求的场景下，或许是一个福音。
延迟 #
延迟是相对于 watermarks 定义的。Watermark(t) 表示事件流的时间已经到达了 t; watermark 之后的时间戳 ≤ t 的任何事件都被称之为延迟事件。
使用 Watermarks #
如果想要使用基于带有事件时间戳的事件流，Flink 需要知道与每个事件相关的时间戳，而且流必须包含 watermark。
动手练习中使用的出租车数据源已经为我们处理了这些详细信息。但是，在您自己的应用程序中，您将必须自己进行处理，这通常是通过实现一个类来实现的，该类从事件中提取时间戳，并根据需要生成 watermarks。最简单的方法是使用 WatermarkStrategy：
DataStream<Event> stream = ...
WatermarkStrategy<Event> strategy = WatermarkStrategy
.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(20))
.withTimestampAssigner((event, timestamp) -> event.timestamp);
DataStream<Event> withTimestampsAndWatermarks =
stream.assignTimestampsAndWatermarks(strategy);