Structured Streaming Watermarks and Output Modes

1.Why stateful queries need a watermark

A stateless streaming query only has to remember which source rows it has already processed. A stateful query also keeps intermediate results across micro-batches, so each batch can update results incrementally instead of recomputing everything. Typical state entities are a time-window aggregation and the unique keys in a stream-stream join. If nothing removes state, it keeps growing, and that leads to memory errors and rising latency. The watermark is what removes it.

You declare a watermark on a streaming DataFrame by naming a timestamp column and a lateness threshold. As data arrives, the state manager tracks the latest timestamp it has seen in that column and processes records that fall within the threshold. The guarantee only goes one way. Records inside the threshold are always processed. Records outside it might still be processed, but that is not guaranteed, so never build logic that relies on late rows being dropped.

from pyspark.sql.functions import window

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

In groupBy() and window(), refer to the event-time column by name (a string or col()) so the event-time marker carries through to the aggregation.

Sources1

2.Choosing thresholds: duration, joins, multiple streams and deduplication

Watermark length trades latency against tolerance for late data. A short watermark keeps less state and writes results sooner, but it tolerates very little late data. A long watermark tolerates more late data, but the query holds more state and waits longer before it writes results.

Stream-stream joins add some rules. For inner joins, Databricks recommends a watermark on each side so old records can be discarded. Without one, every key from both sides is joined on each trigger. For outer joins a watermark is mandatory. Unmatched rows come out with nulls, but only after the lateness threshold passes, because joins only support append mode. Each input can have its own threshold:

(input_stream1.withWatermark("eventTime1", "1 hour")
  .join(
    input_stream2.withWatermark("eventTime2", "2 hours"),
    joinCondition)
)

The engine tracks the maximum event time for each input separately, works out each input's watermark, and then picks one global watermark. By default it uses the minimum, so the query moves at the pace of the slowest stream. If one stream stalls, for example because of an upstream failure, its data is not wrongly marked as late. Setting spark.sql.streaming.multipleWatermarkPolicy to max uses the fastest stream's watermark instead. That lowers latency but drops data from the slower streams.

Deduplication is stateful as well. distinct() keeps every unique record in state, so without a watermark that state grows forever. Structured Streaming guarantees exactly-once processing, but it does not remove duplicates that already exist in the source. To deduplicate on a key such as a GUID, even when duplicates have different event or arrival times, use dropDuplicatesWithinWatermark (Databricks Runtime 13.3 LTS and above). This method requires a watermark. To be sure every duplicate is dropped, set the threshold larger than the maximum time gap between duplicate events.

(streamingDf
  . ? ("eventTime", "10 hours")
  .dropDuplicatesWithinWatermark(["guid"])
)

Sources1

3.Output modes: what a stateful query emits each trigger

Output mode controls which rows a query emits on each trigger. It matters only for stateful queries with aggregations. For stateless queries all modes behave the same. Joins support only append. Output mode has no effect on deduplication, and it does not change how mapGroupsWithState or flatMapGroupsWithState emit rows. If you set no mode, the query runs in append mode.

After that first batch, state holds [2pm, 3pm] at $25 and [3pm, 4pm] at $30. Update mode emits both rows because both changed. Complete mode emits everything. Then $20 at 3:20pm arrives and the watermark moves to 3:05pm. That is past the end of [2pm, 3pm], so append mode now emits that window, which can no longer change. Update mode emits only [3pm, 4pm], whose value went from $30 to $50. Complete mode again emits every row.

Choose the mode based on how downstream consumers use the data. If a consumer takes one action per write, such as a notification service, use append so each record is written only once. If consumers need the freshest values, such as a real-time dashboard or an ML model reading features, use update. Then check that the sink supports the mode. Kafka supports all three. Delta Lake, which backs every Unity Catalog managed table, supports append and complete but not update, so an update-mode write to a managed table fails.

# Append output mode (same as default behavior)
(df.writeStream
  .outputMode("append")
  .toTable("target_table")
)

Sources2

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Update mode works on any sink, including a Delta table.Why is that wrong?

   Delta Lake, and so every Unity Catalog managed table, supports only append and complete. An update-mode write to a managed table fails. Use a merge pattern instead.

   Covered in Output modes: what a stateful query emits each trigger

2. 2.Setting multipleWatermarkPolicy to max is a free way to cut the latency of a multi-stream query.Why is that wrong?

   With max, the global watermark follows the fastest stream, so late-looking data from slower streams is dropped. The default, min, is the safe choice.

   Covered in Choosing thresholds: duration, joins, multiple streams and deduplication

3. 3.Append mode emits an aggregate window as soon as the window's time range ends.Why is that wrong?

   Append mode waits until the watermark (maximum event time minus the delay) passes the window's end, so output is delayed by at least the watermark delay.

   Covered in Output modes: what a stateful query emits each trigger