Structured Streaming · Topic 3 of 3

Streaming Deduplication With and Without a Watermark

Streaming deduplication for the Databricks Spark exam: dropDuplicates state growth, bounding it with a watermark and dropDuplicatesWithinWatermark.

4 min

In a nutshell

dropDuplicates works on streams; distinct() does not.
Without a watermark, all seen keys stay in state, so state grows without bound.
With a watermark, include the event time column in the key columns so old keys can expire.

What it is

Streams often deliver the same record more than once: a producer retries, a source replays after a failure, or a batch is reprocessed. Streaming deduplication removes the repeats so each logical record is kept once.

In a stream Spark can only know a record is a duplicate by remembering the ones it has already seen. That memory is state, and the exam focuses on how to keep it from growing forever.

How it works

Why state is needed

A stream has no end, so Spark cannot compare a record against "all the data". It keeps a store of the keys it has already emitted, inside the query state, and checks each new record against it. The store is saved in the checkpoint, so a restarted query still knows which keys it has seen.

Deduplication without a watermark

dropDuplicates on a streaming DataFrame works on the key columns you give it.

unique = events.dropDuplicates(["event_id"])

Spark stores every key it has seen, in the checkpointed state. Duplicates are caught no matter how far apart they arrive, but the state grows without limit as new keys appear. This is fine for a small key space and a risk for a large or endless one.

Deduplication with a watermark

If the duplicates arrive within a bounded delay, a watermark lets Spark expire old keys. Define the watermark on the event time column and include that column in the key columns.

unique = (events
    .withWatermark("event_time", "10 minutes")
    .dropDuplicates(["event_id", "event_time"]))

Spark drops stored keys once the watermark passes their event time, so state stays bounded. A duplicate that arrives later than the watermark is no longer recognized and may pass through.

This only works when duplicates carry the same event time. If a retry stamps a new event time, the pair event_id and event_time differs and the repeat is not removed.

dropDuplicatesWithinWatermark

dropDuplicatesWithinWatermark (Spark 3.5 and later) solves the changing event time problem. You deduplicate on the identifier alone, and a watermark is still required.

unique = (events
    .withWatermark("event_time", "1 hour")
    .dropDuplicatesWithinWatermark(["event_id"]))

Duplicates are removed if they arrive within the watermark delay. Set the delay longer than the largest time gap between duplicates of the same record.

Whichever form you use, run it on the stream before the sink. Place the call early in the query, so that duplicates are removed before any aggregation counts them twice.

Choosing the approach

Situation	Choice
Few distinct keys, duplicates can arrive any time later	`dropDuplicates` on the id, no watermark
Duplicates arrive close together with the same event time	Watermark plus `dropDuplicates` on id and event time
Duplicates arrive close together but with different event times	Watermark plus `dropDuplicatesWithinWatermark` on the id

How the exam asks about this

"Deduplicate a stream on order_id; memory grows steadily." The state is unbounded; add a watermark and deduplicate with it.
"A retry creates the same order with a new timestamp." Use dropDuplicatesWithinWatermark on order_id.
"Which call removes duplicates from a stream when only an id is the key?" dropDuplicates(["id"]), because distinct() is not supported on streams.

What to remember for the exam

dropDuplicates works on streams; distinct() does not.
Without a watermark, all seen keys stay in state, so state grows without bound.
With a watermark, include the event time column in the key columns so old keys can expire.
Duplicates later than the watermark threshold may no longer be detected.
dropDuplicatesWithinWatermark needs Spark 3.5 or later and a watermark, and deduplicates on the id alone.
Deduplication state is saved in the checkpoint, so it survives a restart.

Watch out

Do not add a watermark and then deduplicate on the id alone with dropDuplicates. The event time column must be part of the key for the watermark to clear state.

Common mistakes

Expecting the watermark to be optional for correctness. Without one, results are correct but state grows forever. Teams hit memory problems weeks later.

Choosing a threshold shorter than the duplicate gap. If duplicates can arrive 30 minutes apart, a 10 minute watermark lets some through. Set the delay to cover the largest gap.

Using a different event time per retry with plain dropDuplicates. The key then differs on every attempt and nothing is removed. This is the case dropDuplicatesWithinWatermark is for.

Check yourself

Why does streaming dropDuplicates without a watermark risk running out of memory?
Spark must remember every key it has ever seen, so the state grows with each new key and is never cleaned up.
What must the key columns include when deduplicating with withWatermark and dropDuplicates?
The event time column the watermark is defined on, so Spark can expire keys as the watermark advances.
When is dropDuplicatesWithinWatermark the right choice?
When duplicates arrive within the watermark delay but carry different event times, so you deduplicate on the id alone.

Topic 3 of 3 in Structured Streaming

Test yourself on Structured Streaming

The full Databricks Associate Developer for Apache Spark question bank has 600 questions, with timed practice exams and an explanation for every answer.

Get Full Access Try Sample Exam

This study guide is independent and is not affiliated with, endorsed by, or sponsored by Databricks. Vendor names, certification names and exam codes are trademarks of their respective owners and are used for identification only.