Watermarks in Apache Flink

Watermarks are fundamental to stream processing. There are many explanations of the concept, which I will not repeat here.

Watermark Characteristics

This is not an official definition, but my understanding is that watermarks have these properties:

• Generated from observed data. Apache Flink supports punctuated and periodic watermark generation, but most users generate them from data already received. This causes some issues discussed below.
• They can only increase. We must therefore consider situations that produce bad watermarks, such as whether to handle timestamps from the future.
• StreamInputProcessor takes the minimum watermark received across channels and passes it downstream. Under data skew, this can contribute to OOM failures.

isWatermarkAligned

The source frequently checks watermark alignment. Ideally, every subtask’s watermark is aligned, but production always brings surprises. Here are two extreme examples.

One stream receives no data for a long time while the others behave normally.

This can happen for many reasons, such as bursty input or upstream filtering. Since a downstream operator uses the minimum watermark from its upstream operators, would one inactive stream prevent the watermark from advancing forever?

No. Flink introduces an idle status for streams. Users configure an IdleTimeout, specifying how long a source can receive no data before being considered idle. After a downstream window operator receives the idle status, it stops using that channel’s previous watermark and calculates its watermark from active channels instead.
As shown above, receiving data causes Source(1) to schedule a callback after the timeout. If no new data arrives during that interval, it sends an idle status downstream and marks itself idle.

Data skew.

Suppose one stream is skewed and the timestamps of its processed events are far behind those of other streams.

In the diagram, if watermarks are set to the received timestamp minus 1, the window’s watermark stays at 0. The window operator accumulates many windows whose state cannot be released, making an OOM likely. There is no good general solution at present; each case needs analysis.

State Of WatermarkOperator

1
stream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[PatternWrapper] {
2
    override def getCurrentWatermark: Watermark = new Watermark(System.currentTimeMillis() - 15000)
3

4
    override def extractTimestamp(element: PatternWrapper, previousElementTimestamp: Long): Long = {
5
      System.currentTimeMillis()
6
    }
7
})

Users can explicitly call this method to generate watermarks. Internally, it creates a WatermarkOperator, which calculates watermarks and sends them downstream.

But if watermarks come from observed data, what happens immediately after recovery, before any data has been sent? We cannot use the current time, because slightly late events would then be filtered out.

This exposes another issue: watermarks are not stored in checkpoints, so WatermarkOperator is stateless. On startup, Flink initializes its watermark to Long.MIN_VALUE. Our architecture is very sensitive to bad data and disallows excessively old historical events, so we made WatermarkOperator stateful. To support changes in parallelism, we restore the minimum watermark across all streams. See the proposals and code changes in FLINK-5601.