[SPARK-24699] [SS]Make watermarks work with Trigger.Once by saving updated watermark to commit log

[tdas]

What changes were proposed in this pull request?

Streaming queries with watermarks do not work with Trigger.Once because of the following.

• Watermark is updated in the driver memory after a batch completes, but it is persisted to checkpoint (in the offset log) only when the next batch is planned
• In trigger.once, the query terminated as soon as one batch has completed. Hence, the updated watermark is never persisted anywhere.

The simple solution is to persist the updated watermark value in the commit log when a batch is marked as completed. Then the next batch, in the next trigger.once run can pick it up from the commit log.

How was this patch tested?

new unit tests

Co-authored-by: Tathagata Das [email protected]
Co-authored-by: c-horn [email protected]

[SparkQA]

Test build #92852 has finished for PR 21746 at commit 7e54a89.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class CommitMetadata(nextBatchWatermarkMs: Long = 0)

[SparkQA]

Test build #93301 has finished for PR 21746 at commit 10b5c2f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #93300 has finished for PR 21746 at commit 584c96e.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[tdas]

@zsxwing

[zsxwing]

LGTM

[HeartSaVioR]

LGTM

[c-horn]

@tdas this will not be included in 2.4.0? as indicated SPARK-24699?

[zsxwing]

@c-horn it's in 2.4.0. I just fixed the ticket.

[c-horn]

Thanks!

[clerxiris]

I have the same problem in PySpark using Spark 2.4.0, that is: Streaming queries with watermarks do not work with .trigger(once=True). When can we expect the same fix for PySpark?

[c-horn]

I believe the current fix only fully processes the on-hand data when window aggregating when you run the Trigger.Once twice. It shouldn't matter that you are using pyspark, it is the same streaming code.

My initial test case was altered such that only the watermark is updated in the first pass, the second pass will process the data that fell before the watermark.

[clerxiris]

Can you provide me with a simple example of the current fix please?

[smrosenberry]

fully processes the on-hand data when window aggregating when you run the Trigger.Once twice

Can someone expound on this comment? Specifically, how do I programmatically "run the Trigger.once twice"?

In its simplest form, I tried this without success:

Dataset dataset;  // input read from a file
DataStreamWriter writer = dataset.writeStream().trigger( Trigger.Once() );

StreamingQuery query = writer.start();
query.awaitTermination();

query = writer.start();
query.awaitTermination();

query.stop();

Any hints would be greatly appreciated...