Mixed format Watermark implementation and how to use it

[majin1102]

Let's talk about Mixed format Watermark implementation currently:

Whatever Flink or spark, a writer writing data whose event time between [A, B] will update Watermark to B - maxAllowLatency if new value > current Watermark. This implementation has two main issues:

1. Spark writing [A, B] has no meaning of data arrival before a fix time. In production scenarios, workflow triggering a Spark job only after the time window of datasource has already passed by and Spark readers does not guarantee reading freshest data. So Spark or batch writers should not update Watermark to trigger downstream jobs. In production scenarios workflow would trigger downstream Spark jobs after upstream Spark job done. Further more, batch workflow triggered by Watermark may not be a good idea.
2. Flink writing [A, B] and updating watermark to C which could be further larger than A, can not mean data before C all arriving, because A which is less than c was just written.

It seems that Watermark in Mixed format currently can not guarantee time arrival using Spark or Flink. How can users use it to trigger downstream jobs for data arrival before a fix time?

I think there are three improvements which could help Watermark effectively used in production scenarios.

1. Make Watermark a more accurate semantics: data arrival guarantee for a fix time window! if Flink commits a snapshot S1 with event time range [A, B], then Watermark should be A for time window [S1, S1], and then Flink commits a snapshot S2 with event range[C, D], Watermark( no mater for S1 or S2) would be minimal(A, C) for time window [S1,S2] . Watermark would be a dynamic corrected value for data arrival guarantee, when users trust a watermark, he always trusts it under a certain time window, for example, I showed transaction list at 2:00AM to find earliest snapshot that guarantee watermark passed 24:00 in last day, finally I found Sn that matched my demand and I use Sn under this fixed time window of (Sn, Sk of 2:00AM).

2. We throw maxAllowLantency which has no meanings for table watermark. compute Watermark as Smallest event time of event time for snapshots and every watermark of certain snapshot could be reduced as new commits happening. Instead of maxAllowLantency we could define a property of WatermarkFixWindow to force a time window before a watermark could be sure.

3. Spark or batch writing does not update Watermark. Watermark is a metrics only for streaming operation.

[zhoujinsong]

I am not sure if I have understood Improvement 1 correctly.
Are you saying that users will find a different snapshot at different times to see the snapshot that meets their watermark needs?
If yes, how should I know if I can trigger the downstream job correctly as it is safe for now but may be not safe later?

[majin1102]

this improvement mainly makes watermark more accurate as time elapsing, so users need to pick a tursted time window for watermark.

[zhoujinsong]

Thanks for your reply, I can understand your meaning now.

But a trusted time window would be hard to decide for me.

Most time, the synchronization task has only a small latency, maybe 5 minutes is safe enough. However, in unusual scenarios, it may be necessary to wait longer, for example, 2 hours.

But when I was using it, I wasn't able to make a good distinction, which had to make me have to wait two hours for all of tasks.

[majin1102]

This is just like offline transfer, when datasource is ready to ETL? God knows, if MySQL replication delays for 5 hours, the entire workflow is wrong, this is just built on a time window trust.

Or think about this, What I need to trust is just how long disorder could last, and before this time window, watermark is robust and some snapshot watermarks is trusted not continue correcting itself.

[zhoujinsong]

However, in some near real-time scenarios where I may need to schedule my downstream tasks every hour or half an hour, it seems that I don't have the opportunity to wait too long to determine the watermark in such scenarios.

[majin1102]

yes, downsteam jobs could do anything without watermark, or only use it as reference.

watermark-fixed-window could be a session property to meet downstream needs.

[zhoujinsong]

I have another question, from your example

if Flink commits a snapshot S1 with event time range [A, B], then Watermark should be A for time window [S1, S1], and then Flink commits a snapshot S2 with event range[C, D], Watermark( no mater for S1 or S2) would be minimal(A, C) for time window [S1,S2]

If A is bigger than C, the watermark would be C at snapshot S2 too, when is the Watermark going to increase?

[zhoujinsong]

Thanks for your reply.

The table watermark helps me determine whether I can start the downstream job.

When should I use the snapshot watermark?

[majin1102]

First table watermark points to some snapshot watermark, so snapshot watermark is a more basic concept.

When I want to read a snapshot which just hits my target watermark(time travel), snapshot watermark could be used by user. for example:

if I want to read 0:00 data without any new data afterwards. I shall use time travel to find a base snapshot less than 0:00 and change snapshot larger than 0:00, and do a merge on read with filtering event time after 0:00 change data.

In my view, using snapshot watermark is a better practice with time travel and table watermark is good to be a metrics and monitored

[zhoujinsong]

Table watermark, which always points to newest trusted snapshot watermark.

Can you describe in more detail what is a trusted snapshot watermark?
Or for example:
S1 commit [A, B], S2 commit [C, D] and A < B < C < D, what is the table watermark?
S1 commit [A, B], S2 commit [C, D] and C < A < B < D, what is the table watermark?

[majin1102]

Ask what is fixed-window first, assume that's 60s which is equal to a commit interval.

S1 commit [A, B], S2 commit [C, D] and A < B < C < D, what is the table watermark?
W(s1) = A, W(s2) = A, if fixed-window is 0, then W(S2) = C,
W(table) = W(S2)

S1 commit [A, B], S2 commit [C, D] and C < A < B < D, what is the table watermark?
W(s1) = C, W(s2) = C
W(table) = W(S2)

[majin1102]

forget about table watermark, only think about snapshot watermark, when a snapshot watermark is good to use

[YesOrNo828]

@majin1102 Thanks for kicking off this improvement. I have some confusion about this discussion.

1. What is the Trusted Snapshot Watermark? You have already answered above, but I don't understand the fixed window very well, should it be exposed to users who own the arctic table?

Flink writing [A, B] and updating watermark to C which could be further larger than A, can not mean data before C all arriving, because A which is less than c was just written.

2. What are the benefits of introducing the concept of the snapshot watermark to solve out-of-sequence data?