一文搞懂FLink WaterMark源码

1. waterMark 源码分析

flink的watermark（水位线）是针对EventTime的盘算，它用来标识应该何时触发窗口盘算，随着数据不停地进入flink步伐，事件的EventTime也将慢慢增大，随之watermark水位线也将逐步上涨，当watermark上涨超过窗口的结束时间，将开始触发窗口盘算。
本文重要分析一下watermark在flink内部的工作原理，包罗watermark的产生、流传、触发盘算的过程。
起首来看看WaterMark的结构,可以看到Watermark的结构非常简单，只有一个时间戳timestamp

1. public final class Watermark implements Serializable {
2.    public Watermark(long timestamp) {
3.       this.timestamp = timestamp;
4.    }
6.    public long getTimestamp() {
7.       return timestamp;
8.    }
10.    public String getFormattedTimestamp() {
11.       return TS_FORMATTER.get().format(new Date(timestamp));
12.    }
13. }

复制代码

1.1.1 设置 waterMark

1. source.assignTimestampsAndWatermarks(WatermarkStrategy
2.         .forBoundedOutOfOrderness(Duration.ofSeconds(5))
3.         .withTimestampAssigner());

复制代码

起首来看assignTimestampsAndWatermarks()方法：

1. public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(
2.       WatermarkStrategy<T> watermarkStrategy) {
4.    final WatermarkStrategy<T> cleanedStrategy = clean(watermarkStrategy);
6.    final TimestampsAndWatermarksOperator<T> operator =
7.       new TimestampsAndWatermarksOperator<>(cleanedStrategy);
9.    // match parallelism to input, to have a 1:1 source -> timestamps/watermarks relationship and chain
10.    final int inputParallelism = getTransformation().getParallelism();
12.    return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator)
13.       .setParallelism(inputParallelism);
14. }

复制代码

可以看到assignTimestampsAndWatermarks()会产生一个新的DataStream，同时会创建一个TimestampsAndWatermarksOperator，看来这个方法和普通的map、filter算子是类似的。
下面附上该代码的执行筹划图，可以看到TimestampsAndWatermarksOperator就是一个普通的operator，可以和filter/map operator进行chain。
1.1.2 打开WatermarksOperator

在Task启动时，会调用operator.open()方法，看看TimestampsAndWatermarksOperator的open方法干了啥

1. public class TimestampsAndWatermarksOperator<T>
2.       extends AbstractStreamOperator<T>
3.       implements OneInputStreamOperator<T, T>, ProcessingTimeCallback {
5.    private static final long serialVersionUID = 1L;
7.    private final WatermarkStrategy<T> watermarkStrategy;
9.    /** The timestamp assigner. */
10.    private transient TimestampAssigner<T> timestampAssigner;
12.    /** The watermark generator, initialized during runtime. */
13.    private transient WatermarkGenerator<T> watermarkGenerator;
15.    /** The watermark output gateway, initialized during runtime. */
16.    private transient WatermarkOutput wmOutput;
18.    /** The interval (in milliseconds) for periodic watermark probes. Initialized during runtime. */
19.    private transient long watermarkInterval;
21.    public TimestampsAndWatermarksOperator(
22.          WatermarkStrategy<T> watermarkStrategy) {
24.       this.watermarkStrategy = checkNotNull(watermarkStrategy);
25.       this.chainingStrategy = ChainingStrategy.ALWAYS;
26.    }
28.    @Override
29.    public void open() throws Exception {
30.       super.open();
32.       timestampAssigner = watermarkStrategy.createTimestampAssigner(this::getMetricGroup);
33.       watermarkGenerator = watermarkStrategy.createWatermarkGenerator(this::getMetricGroup);
35.       // 实例化 WatermarkEmitter 发射器
36.       wmOutput = new WatermarkEmitter(output, getContainingTask().getStreamStatusMaintainer());
38.       // 从配置中获取到 waterMark 生产间隔
39.       watermarkInterval = getExecutionConfig().getAutoWatermarkInterval();
40.       if (watermarkInterval > 0) {
41.          // 根据处理时间注册一个 waterMark 生产间隔闹钟
42.          final long now = getProcessingTimeService().getCurrentProcessingTime();
43.          getProcessingTimeService().registerTimer(now + watermarkInterval, this);
44.       }
45.    }
47.    // 每个数据过来都会调用这个方法
48.    @Override
49.    public void processElement(final StreamRecord<T> element) throws Exception {
50.       final T event = element.getValue();
51.       final long previousTimestamp = element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE;
52.       final long newTimestamp = timestampAssigner.extractTimestamp(event, previousTimestamp);
54.       element.setTimestamp(newTimestamp);
55.       output.collect(element);
56.       // 这里生成 waterMark
57.       watermarkGenerator.onEvent(event, newTimestamp, wmOutput);
58.    }
60.    @Override
61.    public void onProcessingTime(long timestamp) throws Exception {
62.       watermarkGenerator.onPeriodicEmit(wmOutput);
63.       // 注册闹钟
64.       final long now = getProcessingTimeService().getCurrentProcessingTime();
65.       getProcessingTimeService().registerTimer(now + watermarkInterval, this);
66.    }
68.    /**
69.     * Override the base implementation to completely ignore watermarks propagated from
70.     * upstream, except for the "end of time" watermark.
71.     */
72.    @Override
73.    public void processWatermark(org.apache.flink.streaming.api.watermark.Watermark mark) throws Exception {
74.       // if we receive a Long.MAX_VALUE watermark we forward it since it is used
75.       // to signal the end of input and to not block watermark progress downstream
76.       if (mark.getTimestamp() == Long.MAX_VALUE) {
77.          wmOutput.emitWatermark(Watermark.MAX_WATERMARK);
78.       }
79.    }
81.    @Override
82.    public void close() throws Exception {
83.       super.close();
84.       watermarkGenerator.onPeriodicEmit(wmOutput);
85.    }
87.    // ------------------------------------------------------------------------
89.    /**
90.     * Implementation of the {@code WatermarkEmitter}, based on the components
91.     * that are available inside a stream operator.
92.     */
93.     // Watermark 发送器
94.    private static final class WatermarkEmitter implements WatermarkOutput {
96.       private final Output<?> output;
98.       private final StreamStatusMaintainer statusMaintainer;
99.       // 当前的 WaterMark
100.       private long currentWatermark;
102.       private boolean idle;
104.       WatermarkEmitter(Output<?> output, StreamStatusMaintainer statusMaintainer) {
105.          this.output = output;
106.          this.statusMaintainer = statusMaintainer;
107.          this.currentWatermark = Long.MIN_VALUE;
108.       }
110.       @Override
111.       public void emitWatermark(Watermark watermark) {
112.          final long ts = watermark.getTimestamp();
114.          if (ts <= currentWatermark) {
115.             return;
116.          }
118.          currentWatermark = ts;
120.          if (idle) {
121.             idle = false;
122.             statusMaintainer.toggleStreamStatus(StreamStatus.ACTIVE);
123.          }
124.          // 将 waterMark 发送到下游
125.          output.emitWatermark(new org.apache.flink.streaming.api.watermark.Watermark(ts));
126.       }
128.       @Override
129.       public void markIdle() {
130.          idle = true;
131.          statusMaintainer.toggleStreamStatus(StreamStatus.IDLE);
132.       }
133.    }
134. }

复制代码

每条数据来的时间，应该获取到最大的 timeStamp 作为 waterMark

1. public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {
3.    /** The maximum timestamp encountered so far. */
4.    private long maxTimestamp;
6.    /** The maximum out-of-orderness that this watermark generator assumes. */
7.    private final long outOfOrdernessMillis;
9.    /**
10.     * Creates a new watermark generator with the given out-of-orderness bound.
11.     *
12.     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
13.     */
14.    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
15.       checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
16.       checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");
18.       this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
20.       // start so that our lowest watermark would be Long.MIN_VALUE.
21.       this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
22.    }
24.    // ------------------------------------------------------------------------
26.    @Override
27.    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
28.       // 获取最大的 timeStamp作为 WaterMark
29.       maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
30.    }
32.    @Override
33.    public void onPeriodicEmit(WatermarkOutput output) {
34.       // 周期性将 WaterMark 发送出去
35.       output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
36.    }
37. }

复制代码

可以看到open()方法就是注册了一个定时任务，这个定时任务触发的间隔时间就是在步伐里设置的
setAutoWatermarkInterval(interval)这个值，默认是200ms，到达时间之后将会触发target.onProcessingTime(timestamp)

1. // 当闹钟时间到达之后调用这个方法
2. public void onProcessingTime(long timestamp) throws Exception {
3.       // 会调用这个方法
4.       watermarkGenerator.onPeriodicEmit(wmOutput);
5.       final long now = getProcessingTimeService().getCurrentProcessingTime();
6.       // 注册闹钟
7.       getProcessingTimeService().registerTimer(now + watermarkInterval, this);
8.    }

复制代码

终极会调用到WatermarkEmitter发射器的 emitWatermark 方法，会通过 output.emitWatermark 讲当前的 WaterMark 发送出去

1. @Override
2. public void emitWatermark(Watermark watermark) {
3.    final long ts = watermark.getTimestamp();
5.    if (ts <= currentWatermark) {
6.       return;
7.    }
9.    currentWatermark = ts;
11.    if (idle) {
12.       idle = false;
13.       statusMaintainer.toggleStreamStatus(StreamStatus.ACTIVE);
14.    }
16.    output.emitWatermark(new org.apache.flink.streaming.api.watermark.Watermark(ts));
17. }

复制代码

1.1.3 发送 WaterMark

1. //CountingOutput类
2. public void emitWatermark(Watermark mark) {
3.    output.emitWatermark(mark);
4. }
6. //ChainingOutput类
7. public void emitWatermark(Watermark mark) {
8.    try {
9.       // 设置当前 WaterMark
10.       watermarkGauge.setCurrentWatermark(mark.getTimestamp());
11.       if (streamStatusProvider.getStreamStatus().isActive()) {
12.          // 处理 WaterMark
13.          operator.processWatermark(mark);
14.       }
15.    }
16.    catch (Exception e) {
17.       throw new ExceptionInChainedOperatorException(e);
18.    }
19. }
21. //AbstractStreamOperator类
22. public void processWatermark(Watermark mark) throws Exception {
23.     //mapOperator中timeServiceManager == null
24.    if (timeServiceManager != null) {
25.       timeServiceManager.advanceWatermark(mark);
26.    }
27.    output.emitWatermark(mark);
28. }
30. //CountingOutput类
31. public void emitWatermark(Watermark mark) {
32.    output.emitWatermark(mark);
33. }
35. //RecordWriterOutput类
36. public void emitWatermark(Watermark mark) {
37.    watermarkGauge.setCurrentWatermark(mark.getTimestamp());
38.    serializationDelegate.setInstance(mark);
40.    if (streamStatusProvider.getStreamStatus().isActive()) {
41.       try {
42.          // 将 waterMark 信息广播发送
43.          recordWriter.broadcastEmit(serializationDelegate);
44.       } catch (Exception e) {
45.          throw new RuntimeException(e.getMessage(), e);
46.       }
47.    }
48. }
50. //RecordWriter类
51. public void broadcastEmit(T record) throws IOException, InterruptedException {
52.    checkErroneous();
53.    serializer.serializeRecord(record);
55.    boolean pruneAfterCopying = false;
56.    //这里是将watermark广播出去，每一个下游通道都要发送，也就是说会发送到下游的每一个task
57.    for (int channel : broadcastChannels) {
58.       if (copyFromSerializerToTargetChannel(channel)) {
59.          pruneAfterCopying = true;
60.       }
61.    }
63.    // Make sure we don't hold onto the large intermediate serialization buffer for too long
64.    if (pruneAfterCopying) {
65.       serializer.prune();
66.    }
67. }

复制代码

上述代码说明白watermark在本例中的代码调用过程。
在本例中，watermarkOperator和map是chain在一起的，watermarkOperator会将watermark直接通报给map，mapOperator会调用processWatermark()方法来处置惩罚watermark，在mapOperator中timeServiceManager == null，所以mapOperator对watermark不做任务处置惩罚，而是直接将其送达出去。
mapOperator持有的Output是RecordWriterOutput，RecordWriterOutput它会通过RecordWriter将watermark广播到下游的所有通道，即发送给下游的所有task。也就是说，上游的一个task在更新了watermark之后，会将watermark广播给下游的所有task。
3.1.4 下游Task接收并处置惩罚watermark

下游task是一个OneInputStreamTask，通过数据处置惩罚器StreamInputProcessor.processInput()来处置惩罚接收到的数据信息

2. @Override
3. public InputStatus processInput() throws Exception {
4.    // 发送下一个
5.    InputStatus status = input.emitNext(output);
7.    if (status == InputStatus.END_OF_INPUT) {
8.       operatorChain.endHeadOperatorInput(1);
9.    }
11.    return status;
12. }
14. // StreamTaskNetworkInput 类
15. @Override
16. public InputStatus emitNext(DataOutput<T> output) throws Exception {
18.    while (true) {
19.       // get the stream element from the deserializer
20.       if (currentRecordDeserializer != null) {
21.          DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);
22.          if (result.isBufferConsumed()) {
23.             currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
24.             currentRecordDeserializer = null;
25.          }
27.          if (result.isFullRecord()) {
28.             // 处理数据
29.             processElement(deserializationDelegate.getInstance(), output);
30.             return InputStatus.MORE_AVAILABLE;
31.          }
32.       }
34.       Optional<BufferOrEvent> bufferOrEvent = checkpointedInputGate.pollNext();
35.       if (bufferOrEvent.isPresent()) {
36.          // return to the mailbox after receiving a checkpoint barrier to avoid processing of
37.          // data after the barrier before checkpoint is performed for unaligned checkpoint mode
38.          if (bufferOrEvent.get().isEvent() && bufferOrEvent.get().getEvent() instanceof CheckpointBarrier) {
39.             return InputStatus.MORE_AVAILABLE;
40.          }
41.          processBufferOrEvent(bufferOrEvent.get());
42.       } else {
43.          if (checkpointedInputGate.isFinished()) {
44.             checkState(checkpointedInputGate.getAvailableFuture().isDone(), "Finished BarrierHandler should be available");
45.             return InputStatus.END_OF_INPUT;
46.          }
47.          return InputStatus.NOTHING_AVAILABLE;
48.       }
49.    }
50. }
52. private void processElement(StreamElement recordOrMark, DataOutput<T> output) throws Exception {
53.    // 如果是一个数据，那么通过 output 往下游发送数据
54.    if (recordOrMark.isRecord()){
55.       output.emitRecord(recordOrMark.asRecord());
56.    }
57.    // 如果是一个 waterMark，那么通过 output 往下游发送 waterMark
58.    else if (recordOrMark.isWatermark()) {
59.       statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), lastChannel);
60.    } else if (recordOrMark.isLatencyMarker()) {
61.       output.emitLatencyMarker(recordOrMark.asLatencyMarker());
62.    } else if (recordOrMark.isStreamStatus()) {
63.       statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), lastChannel);
64.    } else {
65.       throw new UnsupportedOperationException("Unknown type of StreamElement");
66.    }
67. }

复制代码

StreamInputProcessor会调用statusWatermarkValve.inputWatermark()来处置惩罚接收到watermark信息。看下代码:

1. public void inputWatermark(Watermark watermark, int channelIndex) {
2.    // ignore the input watermark if its input channel, or all input channels are idle (i.e. overall the valve is idle).
3.    if (lastOutputStreamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isActive()) {
4.       long watermarkMillis = watermark.getTimestamp();
6.       // if the input watermark's value is less than the last received watermark for its input channel, ignore it also.
7.       //channelStatuses是当前task对每个inputChannel（也就是每个上游task）的状态信息记录，
8.       //当新的watermark值大于inputChannel的watermark，就会进行调整
9.       if (watermarkMillis > channelStatuses[channelIndex].watermark) {
10.          channelStatuses[channelIndex].watermark = watermarkMillis;
12.          // previously unaligned input channels are now aligned if its watermark has caught up
13.          if (!channelStatuses[channelIndex].isWatermarkAligned && watermarkMillis >= lastOutputWatermark) {
14.             channelStatuses[channelIndex].isWatermarkAligned = true;
15.          }
17.          // now, attempt to find a new min watermark across all aligned channels
18.          //从各个inputChannel的watermark里找到最小的的watermark进行处理
19.          findAndOutputNewMinWatermarkAcrossAlignedChannels();
20.       }
21.    }
22. }
24. private void findAndOutputNewMinWatermarkAcrossAlignedChannels() {
25.    long newMinWatermark = Long.MAX_VALUE;
26.    boolean hasAlignedChannels = false;
28.    // determine new overall watermark by considering only watermark-aligned channels across all channels
29.    //从所有的inputChannels的watermark里找到最小的的watermark
30.    for (InputChannelStatus channelStatus : channelStatuses) {
31.       if (channelStatus.isWatermarkAligned) {
32.          hasAlignedChannels = true;
33.          newMinWatermark = Math.min(channelStatus.watermark, newMinWatermark);
34.       }
35.    }
37.    // we acknowledge and output the new overall watermark if it really is aggregated
38.    // from some remaining aligned channel, and is also larger than the last output watermark
39.    //如果最小的watermark大于之前发送过的watermark，则调用outputHandler进行处理
40.    if (hasAlignedChannels && newMinWatermark > lastOutputWatermark) {
41.       lastOutputWatermark = newMinWatermark;
42.       outputHandler.handleWatermark(new Watermark(lastOutputWatermark));
43.    }
44. }

复制代码

上述代码的大抵实现是，当上游一个task将watermark广播到下游的所有channel（可以明白成下游所有task）之后，下游的task会更新对上游inputChannel记录状态信息中的watermark值，下游每个task都记录这上游所有task的状态值。然后下游task再从所有上游inputChannel（即上游所有task）中选出一个最小值的watermark，如果这个watermark大于最近已经发送的watermark，那么就调用outputHandler对新watermark进行处置惩罚。一般环境下，这个outputHandler就是ForwardingValveOutputHandler。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。