本文介绍了 Watermark 的应用场景，分析了常见的 Watermark 问题及其定位手段。

1 Watermark 应用场景

背景：在现实世界中，网络抖动、链路延时等异常问题，可能导致事件的乱序和延时，即事件产生的顺序和时间不等于事件到达 Flink 的顺序和时间。

Watermark 机制是衡量事件时间的进度，即 Watermark 是 Flink 事件时间特性的基石。因此，我们探讨一下事件时间的应用场景，其主要用于事件时间的窗口 Window（本质是利用事件时间定时器 EventTimer 来实现的）。

对于乱序和迟到的事件，Flink 可以提供收集、等待、丢弃等功能。Watermark 机制可以权衡事件到达和 Window 触发的关系，即确定 Window 的触发时间。

2 Window 窗口不触发问题

Window 窗口实质是事件时间定时器 EventTimer，没有触发 EventTimer 导致 Window 不触发。事件时间定时器 EventTimer 不是实际的定时器，即不是定时器触发的，而是 Watermark触发的（EventTimer 每次注册新的时间间隔，实质是更新当前的 Watermark）。

说明：处理时间定时器 ProcessingTimer 是真实的定时器，当它每次注册新的时间间隔，就是用新的线程进行定时触发。

2.1 Watermark 提取时间失败

Watermark 的 Function 即时间分配器 TimestampAssigner，可以实现 extractTimestamp 接口，从数据中提取事件时间。在缺失事件时间、时间格式错误等异常场景下，提取事件时间失败。

/**
 * Watermark 的时间分配器
 */
public interface TimestampAssigner<T> extends Function {

    /**
     * 从数据中提取事件时间
     */
    long extractTimestamp(T element, long previousElementTimestamp);
}

2.2 Watermark 时间精度下降

由于时间格式变化， timestamp 的时间精度从毫秒级下降到秒级。

例子：时间精度从yyyymmddHHmmssSSS变为yyyymmddHHmmss，窗口大小为10s，如下图所示。
说明：t41、t42、t43是边缘时间戳，即如果第10s的时间是20201020103010，则 t41 是 20201020103010100， t42 是 20201020103010200， t43 是 20201020103010300。精度下降后， t41 为 20201020103010， t42 是 20201020103010， t43 是 20201020103010。

① 时间精度 yyyymmddHHmmssSSS：原来 t41、t42、t43 在 [10,20) 区间内进行窗口计算。
② 时间精度 yyyymmddHHmmss：时间精度下降后， t41、t42、t43 在 [0,10) 区间内进行窗口计算，而[10,20) 区间窗口不触发计算。

Watermark时间精度下降.png

2.3 Watermark 是未来时间

Window 窗口实质是事件时间定时器 EventTimer，而 EventTimer 是 Watermark 触发的，即 EventTimer 每次注册新的时间间隔，实质是更新当前的 Watermark。如果批量数据出现未来时间，EventTimer 会把当前 Watermark 更新为未来时间，即抬高了水位，从而导致后续正常时间的 Watermark 也无法触发 EventTimer。

问题：为什么批量数据出现未来时间，才会导致 EventTimer 更新当前 Watermark 为未来时间？
解答：多并发的场景下，Watermark 是以广播的方式传播到下游算子，而下游算子实例选择最小的 Watermark 更新为当前 low Watermark，并将 low Watermark 发送给下游。

2.4 Watermark 倾斜

（1）概念

Watermark 倾斜实质是 Event Time 倾斜，即 Watermark 中的event Time 时间戳存在较大的差值。

数据倾斜：数据的分布严重不均，造成一部分数据很多，一部分数据很少的局面，即少部分 key 集中大部分数据，导致这些 key 对应的 subtask 负载比其他 subtask 高很多。

（2）场景

在多并发场景下，如果 Flink 作业中的 source 算子（或者 Watermark 算子即 TimestampsAndWatermarksOperator ）发送 Watermark 中的时间戳存在较大的差值，则可能导致 Event Time 倾斜。

例子：如下图所示，Flink 作业消费 Kafka 数据，然后事件时间窗口求和，已知 Kafka 有 2 个 Partition 且数据不均匀，即 Partition-1 的数据远远多于 Partition-2。
解析：
① 由于 Kafka 的 2 个 Partition 数据不均匀，使得 source(1) 和 source(2) 消费的数据也不均匀，导致 source(1) 和 source(2) 产生的 Watermark 有很大的差值。
② 如下图所示，source实例产生 Watermark，同时以广播方式发送 Watermark 到下游，下游的算子实例更新 low Watermark（算子右上角黄色框图）。由于 Window 窗口算子实例有多个输入，它会取最小的 Watermark 为 low Watermark。因此，如下图所示，如果 source(1) 产生 W(29) 、source(2) 产生 W(14)，则 Window 算子实例的 low Watermark 为 W(14)，导致不能触发窗口计算。

多并行下的Watermark.JPG

Window 窗口算子实例取最小的 Watermark 为 low Watermark，其源码如下。WindowOperator 继承 AbstractStreamOperator，并复用 processWatermark、processWatermark1 和 processWatermark2 方法，其中 processWatermark1 和 processWatermark2 处理多个上游发送来的 Watermark。

public abstract class AbstractStreamOperator<OUT>
        implements StreamOperator<OUT>, SetupableStreamOperator<OUT>, CheckpointedStreamOperator, Serializable {
    // 省略...
    public void processWatermark(Watermark mark) throws Exception {
        if (timeServiceManager != null) {
            // 尝试触发 Event Time 定时器
            timeServiceManager.advanceWatermark(mark);
        }
         // 向下游发送 Watermark
        output.emitWatermark(mark);
    }

    public void processWatermark1(Watermark mark) throws Exception {
        input1Watermark = mark.getTimestamp();
         // 获取多个输入的最小 Watermark
        long newMin = Math.min(input1Watermark, input2Watermark);
         // 如果当前最小 Watermark 大于 low Watermrk, 则尝试触发 Event Time 定时器即窗口计算
        if (newMin > combinedWatermark) {
            combinedWatermark = newMin;
            processWatermark(new Watermark(combinedWatermark));
        }
    }

    public void processWatermark2(Watermark mark) throws Exception {
         // 获取多个输入的最小 Watermark
        input2Watermark = mark.getTimestamp();
        long newMin = Math.min(input1Watermark, input2Watermark);
         // 如果当前最小 Watermark 大于 low Watermrk, 则尝试触发 Event Time 定时器即窗口计算
        if (newMin > combinedWatermark) {
            combinedWatermark = newMin;
            processWatermark(new Watermark(combinedWatermark));
        }
    }
    // 省略...
}

3 Watermark 监控

（1）Web UI 监控

Flink Web UI 的作业详情，提供 Watermark 监控，可以观察到算子链的 Low Watermark，如下图所示。

Watermark Web UI监控.JPG

（2）Metrics 监控

Metrics 监控中的 I/O 类别，提供算子的 Watermark 监控信息，包括输入 currentInputWatermark（low Watermark）、currentInputNWatermark、currentOutputWatermark。对比 Web UI 的 Watermark 监控信息，多了 currentInputNWatermark 和 currentOutputWatermark。

① currentInputWatermark：该算子最新接收的 Watermark（如果有多个输入，则取最小值 low Watermark）。
② currentInputNWatermark：第 N 个输入的 Watermark，N 从 1 开始计数。例如从算子实例 operator(1) 接收的为 currentInput1Watermark，从算子实例 operator(2) 接收的为 currentInput2Watermark，以此类推。
③ currentOutputWatermark：该算子最新发送的 Watermark。

【Flink 精选】Watermark 常见问题及其定位思路