Watermark 定义

Watermark 的定义非常简单，它继承了  StreamElement 类，内部只有一个 timestamp 变量。

@PublicEvolvingpublic class Watermark extends StreamElement {    /** The watermark that signifies end-of-event-time. */
    public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);    /** The watermark that signifies is used before any actual watermark has been generated. */
    public static final Watermark UNINITIALIZED = new Watermark(Long.MIN_VALUE);    // ------------------------------------------------------------------------
    /** The timestamp of the watermark in milliseconds. */
    protected final long timestamp;    /** Creates a new watermark with the given timestamp in milliseconds. */
    public Watermark(long timestamp) {        this.timestamp = timestamp;
    }    /** Returns the timestamp associated with this {@link Watermark} in milliseconds. */
    public long getTimestamp() {        return timestamp;
    }    // ------------------------------------------------------------------------
    @Override
    public boolean equals(Object o) {        return this == o
                || o != null
                        && o.getClass() == this.getClass()
                        && ((Watermark) o).timestamp == timestamp;
    }    @Override
    public int hashCode() {        return (int) (timestamp ^ (timestamp >>> 32));
    }    @Override
    public String toString() {        return "Watermark @ " + timestamp;
    }
}

Watermark 处理过程

我们先来回顾一下 Watermark 的生成方法。

SingleOutputStreamOperator<Event> withTimestampsAndWatermarks = source
        .assignTimestampsAndWatermarks(
                WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20))
        );

初始化

在定义 Watermark 的时候，我们调用 assignTimestampsAndWatermarks 方法。

public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(
        WatermarkStrategy<T> watermarkStrategy) {    final WatermarkStrategy<T> cleanedStrategy = clean(watermarkStrategy);    // match parallelism to input, to have a 1:1 source -> timestamps/watermarks relationship
    // and chain
    final int inputParallelism = getTransformation().getParallelism();    final TimestampsAndWatermarksTransformation<T> transformation =            new TimestampsAndWatermarksTransformation<>(                    "Timestamps/Watermarks",
                    inputParallelism,
                    getTransformation(),
                    cleanedStrategy,                    false);
    getExecutionEnvironment(mip.fovkj.com).addOperator(transformation);    return new SingleOutputStreamOperator<>(getExecutionEnvironment(), transformation);
}

这个方法接收了一个 WatermarkStrategy 参数，把它封装到 TimestampsAndWatermarksTransformation 中之后，就添加到 transformations 列表中了。在生成 StreamGraph 的过程中，会调用每个 transformation 的 transform 方法。

通过这个调用链路，创建出了 TimestampsAndWatermarksOperatorFactory，在初始化 StreamTask 时，会调用  TimestampsAndWatermarksOperatorFactory.createStreamOperator 方法来创建 TimestampsAndWatermarksOperator，并调用它的 open 方法。

在这个 open 方法中，主要是生成 timestampAssigner 和 watermarkGenerator。timestampAssigner 是用于提取时间戳，watermarkGenerator 是用于生成 Watermark。

生成完成之后注册了一个定时器，到指定时间后会调用 onProcessingTime 方法。

public void onProcessingTime(long timestamp) throws Exception {
    watermarkGenerator.onPeriodicEmit(wmOutput);    final long now = getProcessingTimeService().getCurrentProcessingTime();
    getProcessingTimeService(nav.fovkj.com).registerTimer(now + watermarkInterval, this);
}

这个方法的逻辑也很简单，先发送创建并发送 Watermark，然后再注册一个定时器。

发送 Watermark

我们以 BoundedOutOfOrdernessWatermarks 为例，它向下游发送了一个 Watermark，时间戳为 maxTimestamp - outOfOrdernessMillis - 1（maxTimestamp 是当前最大的事件时间戳，outOfOrdernessMillis 是我们定义的周期时间毫秒值）。随后在 WatermarkEmitter.emitWatermark 方法中，更新了当前 Watermark 的值。最后 RecordWriterOutput.emitWatermark 则是向下游广播当前的 Watermark。

下游处理

下游处理方法我们从  StreamOneInputProcessor.processInput 入手，先来看具体的调用链路。

在 inputWatermark 方法中，先是对 alignedSubpartitionStatuses 进行调整，alignedSubpartitionStatuses 这个变量主要是用来获取最小的 Watermark。最后调用了  findAndOutputNewMinWatermarkAcrossAlignedSubpartitions 方法。这个方法中，会获取到所有上游最小的 Watermark，如果它大于最近发送的一个 Watermark，就会向下游发送。

public void emitWatermark(Watermark watermark) throws Exception {
    watermarkGauge.setCurrentWatermark(watermark.getTimestamp());
    operator.processWatermark(watermark);

在 tryAdvanceWatermark 方法中如果 Watermark 的时间大于 eventTimeTimersQueue 队列中头节点的时间，那么对 eventTimeTimersQueue 这个队列进行出队操作，这个操作意味着触发了窗口计算。

public boolean tryAdvanceWatermark(        long time, InternalTimeServiceManager.ShouldStopAdvancingFn shouldStopAdvancingFn)
        throws Exception {
    currentWatermark =gov.zuqiu.yj.jrzb1.info time;
    InternalTimer<K, N> timer;    boolean interrupted = false;    while ((timer = eventTimeTimersQueue.peek()) != null
            && timer.getTimestamp() <= time
            && !cancellationContext.isCancelled()
            && !interrupted) {
        keyContext.setCurrentKey(timer.getKey());
        eventTimeTimersQueue.poll();
        triggerTarget.onEventTime(timer);
        taskIOMetricGroup.getNumFiredTimers(gov.gov.mz.zuqb.info).inc();        // Check if we should stop advancing after at least one iteration to guarantee progress
        // and prevent a potential starvation.
        interrupted = shouldStopAdvancingFn.test();
    }    return !interrupted;
}

之后 Watermark 就随着数据流一直到 sink 节点，在 StreamSink 中，支持用户自己实现方法向 sink 中写入 Watermark，除此之外什么也不做。