How do I deduplicate records in a stream?

Summary

• Streaming deduplication needs a unique identifier on each event and state that remembers the identifiers already seen. The design choice is how long to keep that state.
• Spark Structured Streaming uses dropDuplicates, with a watermark to bound state, or dropDuplicatesWithinWatermark when event times differ between duplicates.
• Spark keeps state in executor JVM memory by default. The RocksDB state store provider, available since Spark 3.2, keeps it in native memory and on local disk.
• Flink SQL deduplicates with ROW_NUMBER() and WHERE row_num = 1, and table.exec.state.ttl sets how long the state of a key is retained.
• At the sink, an insert-only Delta Lake merge avoids inserting duplicates, provided the new data is deduplicated within itself.

How do I deduplicate records in a stream?
Give every event a unique identifier and let the engine remember the identifiers it has seen. That memory is state, so the design choice is how long to keep it. The Spark docs say: "The query will store the necessary amount of data from previous records such that it can filter duplicate records." You bound the state with a watermark or a time-to-live, or you accept that it grows.

How does Spark Structured Streaming deduplicate records?

Use dropDuplicates on a unique identifier column, with or without a watermark.

• Without a watermark: "Since there are no bounds on when a duplicate record may arrive, the query stores the data from all the past records as state."
• With a watermark: if there is an upper bound on how late a duplicate may arrive, define a watermark on an event time column and deduplicate on the identifier plus the event time. The query then uses the watermark "to remove old state data from past records that are not expected to get any duplicates anymore."
• With dropDuplicatesWithinWatermark: use it when the event time cannot be part of the identifier because "event times are somehow different for the same records." It requires a watermark with a delay threshold, and the docs encourage setting that threshold "longer than max timestamp differences among duplicated events."

The documented forms, with guid and eventTime columns in streamingDf:

# Without watermark using guid column
streamingDf.dropDuplicates(["guid"])

# With watermark using guid and eventTime columns
streamingDf \
  .withWatermark("eventTime", "10 seconds") \
  .dropDuplicates(["guid", "eventTime"])

# Deduplicate using guid column with watermark based on eventTime column
streamingDf \
  .withWatermark("eventTime", "10 hours") \
  .dropDuplicatesWithinWatermark(["guid"])

Write the deduplicated stream to a Delta table with a checkpoint location:

deduped.writeStream \
  .format("delta") \
  .outputMode("append") \
  .option("checkpointLocation", "/tmp/delta/_checkpoints/") \
  .start("/delta/events")

Where does the deduplication state live?

By default, the HDFS backend state store provider (HDFSBackedStateStore) keeps state in the executors' JVM memory and backs it with files in an HDFS-compatible file system. With millions of keys, the docs say that can cause large garbage collection pauses. The RocksDB state store provider, added in Spark 3.2, manages state "in the native memory and the local disk" instead, and Structured Streaming still saves state changes to your checkpoint location. Enable it by setting spark.sql.streaming.stateStore.providerClass to org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider.

How does Apache Flink deduplicate records?

Flink SQL treats deduplication as a pattern: "Deduplication removes rows that duplicate over a set of columns, keeping only the first one or the last one." You use ROW_NUMBER() with PARTITION BY on the key and ORDER BY on a time attribute (processing time or event time). ASC keeps the first row and DESC keeps the last. The docs warn that "the above pattern must be followed exactly, otherwise the optimizer won't be able to translate the query." The docs example uses an Orders table with a proctime column:

SELECT order_id, user, product, num
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY proctime ASC) AS row_num
  FROM Orders)
WHERE row_num = 1

For state, table.exec.state.ttl "defines for how long the state of a key is retained without being updated before it is removed." After that, a record with the same key "will be treated as if it was the first record with the respective key." Flink ships HashMapStateBackend (the default, objects on the Java heap) and EmbeddedRocksDBStateBackend (RocksDB on local disk), chosen with state.backend.type.

Which approach fits which case?

Approach	State it keeps	Trade-off
Spark dropDuplicates(["guid"])	All past records	State has no upper bound
Spark watermark plus dropDuplicates(["guid", "eventTime"])	Bounded by the watermark	Needs a bound on how late a duplicate arrives, and duplicates must share the event time
Spark dropDuplicatesWithinWatermark(["guid"])	Bounded by the watermark	Deduplicates events that arrive within the delay threshold
Flink SQL ROW_NUMBER()	One entry per key, bounded by table.exec.state.ttl	After the TTL, a repeated key counts as new
Delta Lake insert-only merge	Compares against the target table	The new data must be deduplicated within itself

With the insert-only merge in the Delta Lake docs, "you can avoid inserting the duplicate records":

deltaTable.alias("logs").merge(
  newDedupedLogs.alias("newDedupedLogs"),
  "logs.uniqueId = newDedupedLogs.uniqueId") \
.whenNotMatchedInsertAll() \
.execute()

What does Databricks add?

"RocksDB is the default state store provider in Databricks Runtime 17.3 and above." For earlier runtimes, set spark.sql.streaming.stateStore.providerClass to com.databricks.sql.streaming.state.RocksDBStateStoreProvider. The Databricks docs list dropDuplicatesWithinWatermark for Databricks Runtime 13.3 LTS or above and show the same insert-only merge for Delta tables.

FAQs

Can I deduplicate on more than one column?

Yes. In Spark, pass a list of columns to dropDuplicates. In Flink, PARTITION BY accepts col1[, col2...].

What happens to a duplicate that arrives after the state is gone?

Spark removes state that is not expected to see duplicates anymore, and the docs promise deduplication within the watermark delay threshold. In Flink, an expired key makes the next record count as the first one. If late duplicates matter, deduplicate at the Delta table with merge, which "matches and deduplicates the new data with the existing data in the table".