Study guide
12.4 Streaming Deduplication With and Without Watermarks
Databricks Certified Associate Developer for Apache Spark — Free Study Guide 2026 Exam
Track your progress as you read
Finished sections get a check mark.
1 Chapter 1: Spark Architecture and Execution
1.1 Spark Benefits, Modules, and DataFrame/Dataset Concepts 1.2 Driver, Executors, and Cluster Managers 1.3 Execution Hierarchy: Applications, Jobs, Stages, and Tasks 1.4 Lazy Evaluation, Transformations vs Actions, and Resilient Lineage
2 Chapter 2: Shuffles, Memory and Persistence
2.1 Narrow vs Wide Dependencies and the Shuffle Mechanism 2.2 Memory Management, Fault Tolerance, and Garbage Collection 2.3 Caching, Persistence, and StorageLevels
3 Chapter 3: Spark SQL Sessions, Tables and Views
3.1 SparkSession, Catalogs, Databases, and Namespaces 3.2 Managed vs External Tables and CTAS 3.3 Views: Temporary vs Global Temporary Views
4 Chapter 4: SQL Sources and Stored Layout
4.1 JDBC Reads, Writes, and Parallelism 4.2 Direct File SQL, Save Modes, and Persistent Layout
5 Chapter 5: SQL Queries and Analytics
5.1 SQL Queries, CTEs, Subqueries, and Built-in Expressions 5.2 Built-in SQL Expressions and an Integrated Query 5.3 Window Functions and Complex Joins in Spark SQL 5.4 SQL Join Semantics and Integrated Analytics
6 Chapter 6: DataFrame Construction and Columns
6.1 DataFrame Construction, Schemas, and StructTypes 6.2 Column Operations: Selecting, Renaming, Deriving, and Casting 6.3 Complex Nested Types and User-Defined Functions
7 Chapter 7: DataFrame Validation and Inspection
7.1 Row Operations: Filtering, Sorting, and Deduplication 7.2 Handling Missing and Null Data 7.3 Schema Inspection, Row Iteration, and List Conversion
8 Chapter 8: Aggregations and Dates
8.1 Grouping, Multi-Dimensional Aggregations, and Pivoting 8.2 Dates, Unix Epoch, and Time-Zone Conversions
9 Chapter 9: DataFrame Joins and File IO
9.1 DataFrame Joins and Combining Datasets 9.2 Data Sources, SaveModes, and Partitioning
10 Chapter 10: Shared Variables and Stateful Processing
10.1 Broadcast Variables, Accumulators, and Closures 10.2 Stateless UDFs, Stateful Operators, and StateStores
11 Chapter 11: Troubleshooting and Tuning
11.1 Diagnosing Performance with the Spark UI & Execution Plans 11.2 Data Skew and Partition Tuning: repartition vs coalesce 11.3 Join Optimization and Broadcast Hash Joins
12 Chapter 12: Structured Streaming
12.1 Streaming Architecture, Sources, and Sinks 12.2 Output Modes, Triggers, and Checkpointing 12.3 Event-Time Processing, Watermarking, and Windowing 12.4 Streaming Deduplication With and Without Watermarks
13 Chapter 13: Spark Connect and Deployment
13.1 Spark Connect Architecture & Client-Server Decoupling 13.2 Working with Spark Connect: Connections and Best Practices
14 Chapter 14: Pandas API on Spark
14.1 Pandas API on Apache Spark: Core Architecture and Syntax 14.2 Pandas-on-Spark Operations, Indexing, and Optimization
12.4 Streaming Deduplication With and Without Watermarks
Key Takeaways Explain This Topic AI
Streaming dropDuplicates retains cross-batch state, unlike finite batch deduplication.
Without a watermark, seen-key state can grow indefinitely.
For traditional watermark-based deduplication, including event time in the key allows cleanup but changes equality semantics.
Spark 3.5 dropDuplicatesWithinWatermark can deduplicate stable IDs within a configured event-time delay even when duplicate timestamps differ.
Last updated: October 2026
Why per-batch deduplication is insufficient
A source can resend an event after a timeout, reconnect or upstream replay. If duplicate event_id values arrive in different micro-batches, applying distinct only to a batch DataFrame inside foreachBatch remembers nothing about the previous batch. A streaming deduplication operator instead maintains state across batches. That state must be checkpointed and bounded appropriately for the application.
Choose a key that describes event identity. customer_id would discard a customer's later legitimate events if the desired key is event_id. Conversely, using every payload field can fail to identify the same event when retry metadata or ingestion timestamps differ. Before selecting an API, write down which fields can legitimately change between copies and how long a duplicate may be delayed.
- Without a watermark # events is a streaming DataFrame with event_id and event_time
- unique_forever = events.dropDuplicates(["event_id"])
This remembers seen event IDs across batches and suppresses later duplicates under the query's state history. There is no event-time bound for safe removal of an old key, because another duplicate could arrive arbitrarily far in the future. High-cardinality input can therefore produce growing state, increased checkpoint IO and pressure on memory/native storage. A larger cluster or RocksDB provider can change capacity but does not supply an expiry policy.
Do not assume a trigger interval bounds deduplication memory. A ten-second trigger schedules computation; it does not mean IDs expire after ten seconds. Likewise, a checkpoint path stores progress/state but does not impose business retention or a fixed state-size cap. Monitor retained state and define the intended duplicate horizon.
- Traditional watermark-aware deduplication bounded_exact_pairs = (events
- .withWatermark("event_time", "20 minutes")
- .dropDuplicates(["event_id", "event_time"]))
In this pattern, the watermark timestamp is included in the key so old state can be cleaned up. Equality now means the pair event_id/event_time, not event_id alone. If two copies of an event have the same event ID and the exact same event_time, they match. If the second has a timestamp one second later, the pair differs and it can survive this deduplication. That is a correctness difference, not merely a tuning detail.
Suppose two events are (A, 10:00:00) and (A, 10:00:01). The pair-based pattern treats them as different keys, even though a producer contract may call them retries of one event. Do not remove event_time from an old-API dedup key and assume watermark cleanup still behaves identically. Check the API's documented retention behavior for the supported Spark version.
Spark 3.5: identity within a watermark horizon
Spark 3.5 added dropDuplicatesWithinWatermark for streaming DataFrames with a watermark. It allows deduplication by stable identity without requiring the timestamp itself to be part of the identity key. bounded_ids = (events
- .withWatermark("event_time", "20 minutes")
- .dropDuplicatesWithinWatermark(["event_id"]))
- query = (bounded_ids.writeStream
- .format("parquet")
- .outputMode("append")
- .option("checkpointLocation", "/checkpoints/unique_events")
- .start("/output/unique_events"))
Choose a delay longer than the maximum event-time difference among copies you must reliably suppress, following the API's documented guarantee. This is not unlimited forever-deduplication: once old state expires, a much later copy is outside that retention contract. A bad future timestamp can advance event-time progress unexpectedly; validate event-time quality before choosing a horizon.
The writer path and checkpoint in this example are distinct. A query restart should reuse its own compatible checkpoint to recover identifiers and offsets. A new checkpoint creates a new history, potentially accepting source records previously seen by the old query. Concurrent queries must not share one checkpoint directory.
Watermarks are not universal timestamp filters
Watermark delay is measured relative to observed event-time progress, not the wall-clock age of a record when it reaches a worker. With a latest observed event at 12:45 and delay of twenty minutes, the candidate watermark is 12:25. Advancement is coordinated with micro-batch execution and operator semantics. withWatermark alone does not drop every row older than that timestamp.
For watermark-based operators, inclusion guarantees and late-data handling depend on the API and retained state. Do not generalize a window-aggregation cleanup rule into every deduplication algorithm. A low-traffic stream may not advance its event-time watermark promptly, so state can remain longer in wall-clock time than the nominal delay.
Input identity versus output delivery
Input deduplication suppresses repeated business events while its state is retained. Sink retry protection is a separate requirement. The Kafka sink can produce duplicate output on retry, and foreachBatch is at-least-once unless its writes are made idempotent. Deduplicating before a sink does not automatically make an external insert exactly-once. Use stable keys, batch IDs and appropriate transactional handling at the destination.
For a practical check, test copies in the same batch, in different batches, with identical timestamps, with differing timestamps and outside the chosen horizon. Also test recovery from a compatible checkpoint. Inspect query.lastProgress state-operator metrics and watermark progress rather than assuming a successful startup proves bounded state. Change dedup keys or state schema only through a planned compatible migration. See streaming deduplication and the watermark API .
Choose identity and retention together
- Pattern Identity compared State-retention decision
- Batch dedup inside foreachBatch Keys in that batch only No cross-batch history from this operation
- Streaming dropDuplicates on event_id Stable event ID Unbounded history without cleanup policy
- Watermark plus ordinary ID/time key Exact ID and event-time pair Cleanup can expire old event-time keys
dropDuplicatesWithinWatermark on ID Stable event ID within the API horizon Delay bounds the promised duplicate horizon