Flink (七): DataStream API (四) Watermarks

1. Event Time and Processing Time

1. 1 处理时间（Processing time）

处理时间是指执行相应操作的机器的系统时间。当流处理程序基于处理时间运行时，所有基于时间的操作（如时间窗口）将使用执行相应算子的机器的系统时钟。一个每小时的处理时间窗口将包括在系统时钟指示整点时，所有到达该算子的记录。例如，如果一个应用程序在上午9:15开始运行，第一个每小时的处理时间窗口将包括在9:15到10:00之间处理的事件，接下来的窗口将包括在10:00到11:00之间处理的事件，依此类推。处理时间是最简单的时间概念，不需要在流和机器之间进行协调。它提供了最佳的性能和最低的延迟。然而，在分布式和异步环境中，处理时间无法提供确定性，因为它容易受到记录到达速度（例如来自消息队列）、记录在系统中流动速度以及故障（计划的或其他类型的）等因素的影响。

1.2 事件时间（Event time）

事件时间是每个单独事件发生在其产生设备上的时间。这个时间通常在记录进入 Flink 之前就已嵌入，事件的时间戳可以从每条记录中提取。在事件时间中，时间的进度取决于数据，而不是任何时钟。事件时间程序必须指定如何生成 事件时间水印（Event Time Watermarks）。在理想的情况下，事件时间处理将产生完全一致和确定的结果，无论事件何时到达，或它们的顺序如何。然而，除非事件已知按顺序到达（按时间戳），否则事件时间处理会因为等待乱序事件而产生一定的延迟。由于只能等待有限的时间，这对事件时间应用程序的及时性会造成影响。

假设所有数据都已到达，事件时间操作将按预期行为，并在处理乱序或迟到事件时，或者在重新处理历史数据时，产生正确和一致的结果。例如，一个每小时的事件时间窗口将包含所有事件时间戳落在该小时内的记录，无论它们到达的顺序如何，或者它们何时被处理。

2. Event Time and Watermarks

注意：Flink 实现了许多来自数据流模型（Dataflow Model）的技术。关于事件时间和水印的详细介绍，请参考以下文章：

Tyler Akidau 的《Streaming 101》
《The Dataflow Model》论文

支持事件时间的流处理器需要一种衡量事件时间进度的方法。例如，一个构建按小时划分的窗口的窗口算子，需要在事件时间超过一个小时的结束时得到通知，以便该算子可以关闭当前的窗口。

事件时间可以独立于处理时间。例如，在一个程序中，算子的当前事件时间可能会略微滞后于处理时间（考虑到接收事件的延迟），但两者以相同的速度推进。另一方面，另一个流处理程序可能通过快速推进已经缓存在 Kafka 主题（或其他消息队列）中的一些历史数据，仅用几秒钟就能跨越几周的事件时间。在 Flink 中，用来衡量事件时间进度的机制是水印（Watermarks）。水印作为数据流的一部分流动，并携带一个时间戳 t。一个 Watermark(t) 表示事件时间已经到达时间 t，这意味着在该流中不会再有时间戳 t' <= t 的元素（即事件时间戳小于或等于水印的事件）。

下图展示了带有（逻辑）时间戳的事件流，并且水印与事件一起流动。在这个例子中，事件是按时间戳顺序排列的，这意味着水印只是流中的定期标记。

水印对于乱序流至关重要，正如下图所示，其中事件并未按时间戳排序。一般来说，水印是一个声明，表示在流中的某一点，所有时间戳小于等于某个值的事件都应该已经到达。一旦水印到达算子，算子就可以将其内部的事件时间时钟推进到水印的时间值。

请注意，事件时间是由新创建的流元素继承的，继承自产生这些元素的事件，或者继承自触发这些元素创建的水印。

2.1 Watermarks in Parallel Streams

水印是在源函数处或紧接着源函数之后生成的。每个源函数的并行子任务通常会生成自己的水印。这些水印定义了该并行源的事件时间。随着水印在流处理程序中流动，它们会推进到达算子处的事件时间。每当一个算子推进其事件时间时，它会为下游的后续算子生成一个新的水印。

一些算子会消费多个输入流；例如，union 算子，或者在 keyBy(…) 或 partition(…) 函数之后的算子。此类算子的当前事件时间是其输入流事件时间的最小值。当其输入流更新事件时间时，算子的事件时间也会更新。

下图展示了事件和水印在并行流中流动的示例，以及算子如何追踪事件时间。

3. Lateness

某些元素可能会违反水印条件，即使在 Watermark(t) 发生之后，仍然会有时间戳 t' <= t 的元素到达。事实上，在许多现实世界的应用中，某些元素可能会延迟，这使得无法指定一个时间点，在此之前所有具有特定事件时间戳的元素都已经到达。此外，即使迟到情况可以被处理，水印延迟过多通常也是不理想的，因为这会导致事件时间窗口的延迟过大。

因此，流处理程序可能会明确地预期某些 迟到元素。迟到元素是指在事件时间（由水印表示）已经超过该迟到元素时间戳的时间后到达的元素。对于窗口中如何处理迟到元素，可以参考文章Flink (六)：DataStream API (三) 窗口对于迟到元素处理的一节

4. 生成 Watermark

4.1 Watermark 策略简介

为了使用事件时间语义，Flink 应用程序需要知道事件时间戳对应的字段，意味着数据流中的每个元素都需要拥有可分配的事件时间戳。其通常通过使用 TimestampAssigner API 从元素中的某个字段去访问/提取时间戳。

时间戳的分配与 watermark 的生成是齐头并进的，其可以告诉 Flink 应用程序事件时间的进度。其可以通过指定 WatermarkGenerator 来配置 watermark 的生成方式。

使用 Flink API 时需要设置一个同时包含 TimestampAssigner 和 WatermarkGenerator 的 WatermarkStrategy。WatermarkStrategy 工具类中也提供了许多常用的 watermark 策略，并且用户也可以在某些必要场景下构建自己的 watermark 策略。WatermarkStrategy 接口如下：

public interface WatermarkStrategy<T> 
    extends TimestampAssignerSupplier<T>, WatermarkGeneratorSupplier<T>{

    /**
     * 根据策略实例化一个可分配时间戳的 {@link TimestampAssigner}。
     */
    @Override
    TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);

    /**
     * 根据策略实例化一个 watermark 生成器。
     */
    @Override
    WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

如上所述，通常情况下，你不用实现此接口，而是可以使用 WatermarkStrategy 工具类中通用的 watermark 策略，或者可以使用这个工具类将自定义的 TimestampAssigner 与 WatermarkGenerator 进行绑定。例如，你想要要使用有界无序（bounded-out-of-orderness）watermark 生成器和一个 lambda 表达式作为时间戳分配器，那么可以按照如下方式实现：

WatermarkStrategy
        .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
        .withTimestampAssigner((event, timestamp) -> event.f0);

其中 TimestampAssigner 的设置与否是可选的，大多数情况下，可以不用去特别指定。例如，当使用 Kafka 或 Kinesis 数据源时，你可以直接从 Kafka/Kinesis 数据源记录中获取到时间戳。

4.2 使用 Watermark 策略

WatermarkStrategy 可以在 Flink 应用程序中的两处使用，第一种是直接在数据源上使用，第二种是直接在非数据源的操作之后使用。

第一种方式相比会更好，因为数据源可以利用 watermark 生成逻辑中有关分片/分区（shards/partitions/splits）的信息。使用这种方式，数据源通常可以更精准地跟踪 watermark，整体 watermark 生成将更精确。直接在源上指定 WatermarkStrategy 意味着你必须使用特定数据源接口，仅当无法直接在数据源上设置策略时，才应该使用第二种方式（在任意转换操作之后设置 WatermarkStrategy）：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<MyEvent> stream = env.readFile(
        myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100,
        FilePathFilter.createDefaultFilter(), typeInfo);

DataStream<MyEvent> withTimestampsAndWatermarks = stream
        .filter( event -> event.severity() == WARNING )
        .assignTimestampsAndWatermarks(<watermark strategy>);

withTimestampsAndWatermarks
        .keyBy( (event) -> event.getGroup() )
        .window(TumblingEventTimeWindows.of(Time.seconds(10)))
        .reduce( (a, b) -> a.add(b) )
        .addSink(...);

使用 WatermarkStrategy 去获取流并生成带有时间戳的元素和 watermark 的新流时，如果原始流已经具有时间戳或 watermark，则新指定的时间戳分配器将覆盖原有的时间戳和 watermark。

4.3 处理空闲数据源

如果数据源中的某一个分区/分片在一段时间内未发送事件数据，则意味着 WatermarkGenerator 也不会获得任何新数据去生成 watermark。我们称这类数据源为空闲输入或空闲源。在这种情况下，当某些其他分区仍然发送事件数据的时候就会出现问题。由于下游算子 watermark 的计算方式是取所有不同的上游并行数据源 watermark 的最小值，则其 watermark 将不会发生变化。

为了解决这个问题，你可以使用 WatermarkStrategy 来检测空闲输入并将其标记为空闲状态。WatermarkStrategy 为此提供了一个工具接口：

WatermarkStrategy
        .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
        .withIdleness(Duration.ofMinutes(1));

4.4 Watermark alignment

在前一段中，我们讨论了当分片/分区/分片或源处于空闲状态并可能导致水印停滞的情况。另一方面，某个分片/分区/分片或源可能处理记录的速度非常快，从而使其水印相对其他部分更快地推进。单独来看，这本身并不是一个问题。然而，对于那些使用水印来发送数据的下游算子来说，这可能会成为一个问题。

在这种情况下，与空闲源相反，某些下游算子（例如带窗口的连接或聚合操作）的水印可以推进。然而，这些算子可能需要缓冲来自快速输入的过多数据，因为所有输入流的最小水印会被滞后的输入流拖慢。因此，来自快速输入的所有记录必须缓存在下游算子的状态中，这可能会导致算子状态的不可控增长。

为了解决这个问题，可以启用水印对齐，这样可以确保没有源/分片/分区/分片的水印推进得过快。您可以为每个源单独启用对齐：

WatermarkStrategy
        .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
        .withWatermarkAlignment("alignment-group-1", Duration.ofSeconds(20), Duration.ofSeconds(1));

在启用水印对齐时，您需要告诉 Flink 哪个组应该包含该源。您可以通过提供一个标签（例如：alignment-group-1）来将所有共享该标签的源绑定在一起。此外，您还需要告诉 Flink 所有属于该组的源的当前最小水印之间的最大偏移量。第三个参数描述了当前最大水印应该多久更新一次。频繁更新的缺点是会产生更多的 RPC 消息在任务管理器（TM）和作业管理器（JM）之间传递。

为了实现水印对齐，Flink 会暂停消费来自产生过早水印的源/任务的记录，同时继续从其他源/任务读取记录，这些记录能够推动合并水印向前推进，从而解除较快源的阻塞。

4.5 自定义 WatermarkGenerator

TimestampAssigner 是一个可以从事件数据中提取时间戳字段的简单函数，我们无需详细查看其实现。但是 WatermarkGenerator 的编写相对就要复杂一些了，WatermarkGenerator 接口代码如下：

/**
 * {@code WatermarkGenerator} 可以基于事件或者周期性的生成 watermark。
 *
 * <p><b>注意：</b>  WatermarkGenerator 将以前互相独立的 {@code AssignerWithPunctuatedWatermarks} 
 * 和 {@code AssignerWithPeriodicWatermarks} 一同包含了进来。
 */
@Public
public interface WatermarkGenerator<T> {

    /**
     * 每来一条事件数据调用一次，可以检查或者记录事件的时间戳，或者也可以基于事件数据本身去生成 watermark。
     */
    void onEvent(T event, long eventTimestamp, WatermarkOutput output);

    /**
     * 周期性的调用，也许会生成新的 watermark，也许不会。
     *
     * <p>调用此方法生成 watermark 的间隔时间由 {@link ExecutionConfig#getAutoWatermarkInterval()} 决定。
     */
    void onPeriodicEmit(WatermarkOutput output);
}

watermark 的生成方式本质上是有两种：周期性生成和标记生成。

周期性生成器通常通过 onEvent() 观察传入的事件数据，然后在框架调用 onPeriodicEmit() 时发出 watermark。

标记生成器将查看 onEvent() 中的事件数据，并等待检查在流中携带 watermark 的特殊标记事件或打点数据。当获取到这些事件数据时，它将立即发出 watermark。通常情况下，标记生成器不会通过 onPeriodicEmit() 发出 watermark。

4.5.1 自定义周期性 Watermark 生成器

周期性生成器会观察流事件数据并定期生成 watermark（其生成可能取决于流数据，或者完全基于处理时间）。

生成 watermark 的时间间隔（每 n 毫秒）可以通过 ExecutionConfig.setAutoWatermarkInterval(...) 指定。每次都会调用生成器的 onPeriodicEmit() 方法，如果返回的 watermark 非空且值大于前一个 watermark，则将发出新的 watermark。

如下是两个使用周期性 watermark 生成器的简单示例。注意：Flink 已经附带了 BoundedOutOfOrdernessWatermarks，它实现了 WatermarkGenerator，其工作原理与下面的 BoundedOutOfOrdernessGenerator 相似

/**
 * 该 watermark 生成器可以覆盖的场景是：数据源在一定程度上乱序。
 * 即某个最新到达的时间戳为 t 的元素将在最早到达的时间戳为 t 的元素之后最多 n 毫秒到达。
 */
public class BoundedOutOfOrdernessGenerator implements WatermarkGenerator<MyEvent> {

    private final long maxOutOfOrderness = 3500; // 3.5 秒

    private long currentMaxTimestamp;

    @Override
    public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 发出的 watermark = 当前最大时间戳 - 最大乱序时间
        output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
    }

}

/**
 * 该生成器生成的 watermark 滞后于处理时间固定量。它假定元素会在有限延迟后到达 Flink。
 */
public class TimeLagWatermarkGenerator implements WatermarkGenerator<MyEvent> {

    private final long maxTimeLag = 5000; // 5 秒

    @Override
    public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        // 处理时间场景下不需要实现
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(System.currentTimeMillis() - maxTimeLag));
    }
}

4.5.2 自定义标记 Watermark 生成器

标记 watermark 生成器观察流事件数据并在获取到带有 watermark 信息的特殊事件元素时发出 watermark。

如下是实现标记生成器的方法，当事件带有某个指定标记时，该生成器就会发出 watermark：

public class PunctuatedAssigner implements WatermarkGenerator<MyEvent> {

    @Override
    public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        if (event.hasWatermarkMarker()) {
            output.emitWatermark(new Watermark(event.getWatermarkTimestamp()));
        }
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // onEvent 中已经实现
    }
}

4.6 Watermark 策略与 Kafka 连接器

当使用 Apache Kafka 连接器作为数据源时，每个 Kafka 分区可能有一个简单的事件时间模式（递增的时间戳或有界无序）。然而，当使用 Kafka 数据源时，多个分区常常并行使用，因此交错来自各个分区的事件数据就会破坏每个分区的事件时间模式（这是 Kafka 消费客户端所固有的）。

在这种情况下，你可以使用 Flink 中可识别 Kafka 分区的 watermark 生成机制。使用此特性，将在 Kafka 消费端内部针对每个 Kafka 分区生成 watermark，并且不同分区 watermark 的合并方式与在数据流 shuffle 时的合并方式相同。

例如，如果每个 Kafka 分区中的事件时间戳严格递增，则使用单调递增时间戳分配器按分区生成的 watermark 将生成完美的全局 watermark。注意，我们在示例中未使用 TimestampAssigner，而是使用了 Kafka 记录自身的时间戳。

下图展示了如何使用单 kafka 分区 watermark 生成机制，以及在这种情况下 watermark 如何通过 dataflow 传播。

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("my-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> stream = env.fromSource(
    kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20)), "mySource");

4.7 算子处理 Watermark 的方式

一般情况下，在将 watermark 转发到下游之前，需要算子对其进行触发的事件完全进行处理。例如，WindowOperator 将首先计算该 watermark 触发的所有窗口数据，当且仅当由此 watermark 触发计算进而生成的所有数据被转发到下游之后，其才会被发送到下游。换句话说，由于此 watermark 的出现而产生的所有数据元素都将在此 watermark 之前发出。

相同的规则也适用于 TwoInputStreamOperator。但是，在这种情况下，算子当前的 watermark 会取其两个输入的最小值。

5. 内置 Watermark 生成器

5.1 单调递增时间戳分配器

周期性 watermark 生成方式的一个最简单特例就是你给定的数据源中数据的时间戳升序出现。在这种情况下，当前时间戳就可以充当 watermark，因为后续到达数据的时间戳不会比当前的小。

注意：在 Flink 应用程序中，如果是并行数据源，则只要求并行数据源中的每个单分区数据源任务时间戳递增。例如，设置每一个并行数据源实例都只读取一个 Kafka 分区，则时间戳只需在每个 Kafka 分区内递增即可。Flink 的 watermark 合并机制会在并行数据流进行分发（shuffle）、联合（union）、连接（connect）或合并（merge）时生成正确的 watermark。

WatermarkStrategy.forMonotonousTimestamps();

5.2 数据之间存在最大固定延迟的时间戳分配器

另一个周期性 watermark 生成的典型例子是，watermark 滞后于数据流中最大（事件时间）时间戳一个固定的时间量。该示例可以覆盖的场景是你预先知道数据流中的数据可能遇到的最大延迟，例如，在测试场景下创建了一个自定义数据源，并且这个数据源的产生的数据的时间戳在一个固定范围之内。Flink 针对上述场景提供了 boundedOutfordernessWatermarks 生成器，该生成器将 maxOutOfOrderness 作为参数，该参数代表在计算给定窗口的结果时，允许元素被忽略计算之前延迟到达的最长时间。其中延迟时长就等于 t - t_w ，其中 t 代表元素的（事件时间）时间戳，t_w 代表前一个 watermark 对应的（事件时间）时间戳。如果 lateness > 0，则认为该元素迟到了，并且在计算相应窗口的结果时默认会被忽略

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10));