Apache Beam setting custom Event timestamp and watermark

Question

I'm reading from Kafka using Apache Beam. I want to use the event timestamp from the payload instead of the default LogAppendTime due to out of order writes to Kafka.

I have seen a solution in Apache Beam: Error assigning event time using Withtimestamp

How do we define the watermark? Considering it should be monotonic.

In the method

getWatermark(TimestampPolicy.PartitionContext ctx)

How do we determine the watermark when there is an event which is hours late.

This seems to be a very common scenario of out of order writes and event time windowing. I did not see much literature on how this.

Reading the comments of code makes me feel that watermark is a builtin machism in IO/Beam: github.com/apache/beam/blob/master/sdks/java/io/kafka/src/main/… — Rui Wang
The builtin uses LogAppendTime which is monotonic. It will always progress forward even in case out of order writes — PK109

Robert Fišer Robert Fišer · Accepted Answer · 2019-09-10T11:38:31

You may extract timestamp from data using ParDo and outputWithTimestamp like its described here https://beam.apache.org/documentation/programming-guide/#adding-timestamps-to-a-pcollections-elements

Apache Beam setting custom Event timestamp and watermark

1 Answers