Hello @ksenija ,

Greetings!

Streaming uses watermarks to control the threshold for how long to continue processing updates for a given state entity. Common examples of state entities include:

• Aggregations over a time window.

• Unique keys in a join between two streams.

When you declare a watermark, you specify a timestamp field and a watermark threshold on a streaming DataFrame. As new data arrives, the state manager tracks the most recent timestamp in the specified field and processes all records within the lateness threshold.

The following example applies a 10 minute watermark threshold to a windowed count:

from pyspark.sql.functions import window

(df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window("event_time", "5 minutes"),
    "id")
  .count()
)

In this example:

• The event_time column is used to define a 10 minute watermark and a 5 minute tumbling window.

• A count is collected for each id observed for each non-overlapping 5 minute windows.

• State information is maintained for each count until the end of window is 10 minutes older than the latest observed event_time.

You can read more about watermark here: https://docs.databricks.com/en/structured-streaming/watermarks.html

https://www.databricks.com/blog/feature-deep-dive-watermarking-apache-spark-structured-streaming

Regards,

Ravi