Your watermarked dedup is still hoarding state

Two conditions must be true

First, apply withWatermark before dropDuplicates. Every official example uses this order, and the Databricks documentation says it plainly: # Apply watermark before distinct operation. Reverse the chain and the watermark never reaches the dedup operator. Spark documents the consequence under “Without watermark”: “the query stores the data from all the past records as state.” No error is documented for the reversed order. The dangerous outcome is a query that runs while state keeps accumulating.

Second, put the event-time column in the dedup key list. Correct order alone is not enough for dropDuplicates. In the Databricks Community deep-dive's documented run, the chain was .withWatermark("event_time", "5 minutes").dropDuplicates(["event_id"]): the order was right, but event_time was missing from the keys. After the watermark advanced to 10:02, keys A and B remained in the state store. The write-up calls this global deduplication: the state store grows continuously, potentially causing out-of-memory issues.

The A/B result above is the outcome of that documented run, not a deterministic per-record arrival guarantee. Watermarks are threshold semantics: they control when state is eligible to be evicted as the engine's watermark advances.

Setup

events = (
    spark.readStream
         .format("delta")
         .table("main.bronze.events")
         .select("event_time", "event_id", "payload")
)

REVERSED — the watermark never reaches dedup

reversed_order = (
    events
      .dropDuplicates(["event_id", "event_time"])
      .withWatermark("event_time", "5 minutes")
)

MISSING-KEY — correct order, still global

global_dedup = (
    events
      .withWatermark("event_time", "5 minutes")
      .dropDuplicates(["event_id"])
)

FIXED — Spark's documented key pattern

deduplicated = (
    streamingDf
      .withWatermark("eventTime", "10 seconds")
      .dropDuplicates(["guid", "eventTime"])
)

SAFER ALTERNATIVE — bound duplicate lateness directly

bounded_dedup = (
    streamingDf
      .withWatermark("eventTime", "10 seconds")
      .dropDuplicatesWithinWatermark(["guid"])
)

dropDuplicatesWithinWatermark requires Databricks Runtime 13.3 LTS or above. It guarantees deduplication only for duplicates arriving within the watermark threshold. A key is removed when the watermark reaches or passes event_time + delay. Set the threshold greater than the maximum timestamp difference between duplicate events. It needs a watermark to work at all.

Verify it yourself

Use the state-store reader from the deep-dive to list stored keys. Run it against the checkpoint for your query, advance the input past the watermark, and confirm old keys disappear.

state_keys = (
    spark.read
         .format("statestore")
         .load("<checkpointLocation>")
)

state_keys.show(truncate=False)

1. Check the chain: withWatermark is applied before dropDuplicates.
2. Check the key: the event-time column is in the dedup key list — or use dropDuplicatesWithinWatermark on Databricks Runtime 13.3 LTS+.
3. Check the state: inspect the state store and confirm old keys are actually evicted as the watermark advances.

Use this when

Deduplicating a stream where duplicates arrive within a known, bounded delay. Choose a watermark threshold that exceeds the maximum timestamp difference between duplicate events, then verify eviction in state.

Avoid this when

Duplicates can arrive arbitrarily late. Plain global dropDuplicates without a watermark is the documented option, but it stores all past records as state. Treat that as a memory budget, not a config flag. Do not use dropDuplicatesWithinWatermark below Databricks Runtime 13.3 LTS.