50、Flink 数据源的事件时间和水印详解

事件时间和水印

a）概述

Source 的实现需要完成一部分事件时间分配和水印生成的工作，离开 SourceReader 的事件流需要具有事件时间戳，并且（在流执行期间）包含水印。

旧版 SourceFunction 的应用通常在之后的单独的一步中通过 stream.assignTimestampsAndWatermarks(WatermarkStrategy) 生成时间戳和水印；这个函数不应该与新的 Sources 一起使用，因为此时时间戳应该已经被分配了，而且该函数会覆盖掉之前的分片（split-aware）水印。

b）API

在 DataStream API 创建期间，WatermarkStrategy 会被传递给 Source，并同时创建 TimestampAssigner 和 WatermarkGenerator。

environment.fromSource(
    Source source,
    WatermarkStrategy timestampsAndWatermarks,
    String sourceName);

TimestampAssigner 和 WatermarkGenerator 作为 ReaderOutput（或 SourceOutput）的一部分透明地运行，因此 Source 实现者不必实现任何时间戳提取和水印生成的代码。

c）事件时间戳

事件时间戳的分配分为以下两步：

• SourceReader 通过调用 SourceOutput.collect(event, timestamp) 将 Source 记录的时间戳添加到事件中。 该实现只能用于含有记录并且拥有时间戳特性的数据源，例如 Kafka、Kinesis、Pulsar 或 Pravega。 因此，记录中不带有时间戳特性的数据源（如文件）也就无法实现这一步了。 此步骤是 Source 连接器实现的一部分，不由使用 Source 的应用程序进行参数化设定。
• 由应用程序配置的 TimestampAssigner 分配最终的时间戳。 TimestampAssigner 会查看原始的 Source 记录的时间戳和事件。分配器可以直接使用 Source 记录的时间戳或者访问事件的某个字段获得最终的事件时间戳。

这种分两步的方法使用户既可以引用 Source 系统中的时间戳，也可以引用事件数据中的时间戳作为事件时间戳。

注意： 当使用没有 Source 记录的时间戳的数据源（如文件）并选择 Source 记录的时间戳作为最终的事件时间戳时，默认的事件时间戳等于 LONG_MIN (=-9,223,372,036,854,775,808)。

d）水印生成

水印生成器仅在流执行期间会被激活。批处理执行则会停用水印生成器，则下文所述的所有相关操作实际上都变为无操作。

数据 Source API 支持每个分片单独运行水印生成器。这使得 Flink 可以分别观察每个分片的事件时间进度，这对于正确处理事件时间偏差和防止空闲分区阻碍整个应用程序的事件时间进度来说是很重要的。

使用 SplitReader API 实现源连接器时，将自动进行处理。所有基于 SplitReader API 的实现都具有开箱即用的分片水印。

为了保证更底层的 SourceReader API 可以使用每个分片的水印生成，必须将不同分片的事件输送到不同的输出（outputs）中：局部分片（Split-local） SourceOutputs。通过 createOutputForSplit(splitId) 和 releaseOutputForSplit(splitId) 方法，可以在总 ReaderOutput 上创建并发布局部分片输出。

e）拆分级别的水印对齐

尽管 source operator 的水印对齐由 Flink 运行时处理，但 source 需要额外实现 SourceReader#pauseOrResumeSplits 和SplitReader#pouseOrResueSplits，以实现拆分级别的水印对齐。

当有多个 splits 分配给一个 source reader 时，拆分级别的水印对齐非常有用；默认情况下，当分配了多个 splits，pipeline.watermark-alignment.allow-unaligned-source-splits 设置为false，并且分割超过 WatermarkStrategy 配置的水印对齐阈值时，这些实现将抛出 UnsupportedOperationException。

SourceReaderBase 包含 SourceReader#pauseOrResumeSplits 的实现，因此继承 sources 只需要实现SplitReader#pouseOrResueSplits。