Deduplication strategies for Spark & Data Eng developers

Introduction

Duplicate records are one of the most common data quality issues you'll encounter in production pipelines, and they compound silently until your reports are wrong. Deduplication strategies for Spark & Data Eng developers range from simple one-liners to sophisticated window-based approaches, and choosing the right one depends heavily on your data model, source systems, and SLA requirements.

In this guide, we'll walk through the practical techniques we use to remove duplicates at scale, from leveraging business keys to handling streaming scenarios with watermarks. Each approach has trade-offs in terms of correctness, performance, and operational complexity.

Understanding Why Duplicates Occur

Before we jump into solutions, it's worth understanding the root causes. Duplicates typically arise from:

• Source system retries (API calls that succeed but report failure)
• Multiple data loads in a single batch window
• ETL jobs that restart and re-process overlapping time ranges
• Schema changes that break deduplication logic downstream
• Joins that explode row counts unintentionally

Getting ahead of duplicates means thinking about deduplication at multiple layers: source ingestion, transformation, and consumption.

dropDuplicates on Business Keys

The simplest approach is Spark's dropDuplicates() method, which removes rows where all specified columns match. This works well when you have a clear business key—a set of columns that uniquely identify a record in your domain.

For an e-commerce system, a business key might be order_id. For a user event stream, it could be (user_id, event_type, event_timestamp). When duplicates appear, Spark keeps one arbitrary row and discards the rest.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("dedup").getOrCreate()

# Sample data with duplicates
data = [
    ("order_1", "alice", 100.0),
    ("order_1", "alice", 100.0),
    ("order_2", "bob", 250.0),
    ("order_2", "bob", 250.0),
    ("order_3", "charlie", 75.0),
]

df = spark.createDataFrame(data, ["order_id", "customer_name", "amount"])

# Simple deduplication on business key
deduped_df = df.dropDuplicates(["order_id"])
deduped_df.show()
# Output: 3 rows (one per order_id)

The strength of dropDuplicates() is its simplicity and Spark's optimized shuffle-based implementation. The weakness is that it doesn't let you control which duplicate row to keep. If you need the latest timestamp or the row with the most complete data, you need a different approach.

In production, we typically use dropDuplicates() for idempotent deduplication in batch jobs where all duplicate rows are truly identical and you don't care which one survives.

Window Functions to Keep Latest

When duplicates have subtle differences—like multiple records for the same transaction with different timestamps or statuses—you need finer control. Window functions with row_number() or rank() let you rank duplicates and keep only the one you want.

A common pattern is keeping the most recent record based on a timestamp column, which works well for slowly-changing dimensions or late-arriving data.

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, desc

# Sample data with duplicates at different times
data = [
    ("order_1", "alice", 100.0, "2024-01-01 10:00:00"),
    ("order_1", "alice", 105.0, "2024-01-01 10:05:00"),  # Later update
    ("order_2", "bob", 250.0, "2024-01-01 09:00:00"),
    ("order_2", "bob", 250.0, "2024-01-01 09:30:00"),
    ("order_3", "charlie", 75.0, "2024-01-01 11:00:00"),
]

df = spark.createDataFrame(
    data, 
    ["order_id", "customer_name", "amount", "updated_at"]
)

# Define window: partition by business key, order by timestamp desc
window_spec = Window.partitionBy("order_id").orderBy(desc("updated_at"))

# Rank within each partition; row_number gives 1 to the first (most recent)
deduped_df = (
    df.withColumn("rn", row_number().over(window_spec))
    .filter("rn = 1")
    .drop("rn")
)

deduped_df.show()
# Output: 3 rows, each with the latest amount and timestamp per order_id

This pattern is powerful because you can rank by multiple criteria: order by timestamp descending, then by a completeness score, then by data quality flags. You have full control over which duplicate survives.

The trade-off is performance: window functions require a shuffle and can be slower than simple dropDuplicates() on large datasets. Use this when correctness outweighs raw speed, which is most of the time.

Leveraging Event IDs from Source Systems

The best deduplication strategy is prevention. Many modern systems—payment gateways, message brokers, SaaS APIs—provide idempotency keys or event IDs that prevent duplicates at the source.

If your upstream systems issue unique IDs for each logical event (not each transmission), you can build idempotent pipelines by storing processed IDs in a state store and skipping anything already seen.

from pyspark.sql.functions import col, broadcast

# Incoming events
events_data = [
    ("evt_001", "user_123", "purchase", 99.99),
    ("evt_001", "user_123", "purchase", 99.99),  # Duplicate
    ("evt_002", "user_456", "view", 0.0),
    ("evt_003", "user_123", "purchase", 150.0),
]

events_df = spark.createDataFrame(
    events_data,
    ["event_id", "user_id", "event_type", "value"]
)

# Previously processed events (from state store or checkpoint)
processed_ids = spark.createDataFrame(
    [("evt_001",), ("evt_004",)],
    ["event_id"]
)

# Anti-join: keep only events NOT in processed state
new_events_df = events_df.join(
    broadcast(processed_ids),
    on="event_id",
    how="left_anti"
)

new_events_df.show()
# Output: 2 rows (evt_002 and evt_003)

This approach is especially valuable in microservices and event-driven architectures. By relying on source-level IDs, you avoid the brittleness of business-key-only deduplication and gain audit trails for free.

Store processed event IDs in Redis, DynamoDB, or a state table partitioned by time window so you can query recent history efficiently without scanning years of data.

Deduplication in Streaming with Watermarks

Streaming pipelines introduce latency: records can arrive out of order, and older data can show up after newer data has already been processed. Standard deduplication approaches don't work when state is distributed across micro-batches.

We use watermarks to define a cutoff point: once a watermark passes a record's timestamp, we assume no older data will arrive for that time window. Combined with state stores, this lets us deduplicate correctly even with late arrivals.

from pyspark.sql.functions import col, from_json, from_unixtime
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType

schema = StructType([
    StructField("event_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("timestamp", LongType()),
])

# Read from Kafka
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .load()

# Parse JSON and add event time
parsed_df = kafka_df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*").select(
    col("event_id"),
    col("user_id"),
    col("amount"),
    from_unixtime(col("timestamp")).cast("timestamp").alias("event_time")
)

# Set watermark: allow 10 minutes of late data
watermarked_df = parsed_df.withWatermark("event_time", "10 minutes")

# Deduplicate using event_id, grouped by a time window
from pyspark.sql.functions import window
deduped_stream = watermarked_df \
    .groupBy("event_id") \
    .agg({"*": "first"})

# Write to console (or any sink)
query = deduped_stream.writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

query.awaitTermination()

The key insight is that watermarks don't deduplicate by themselves; they tell Spark when it's safe to finalize state. You still need to group or use window functions to actually deduplicate.

For truly stateful deduplication in streaming, consider using Spark Structured Streaming with mapGroupsWithState(), which maintains a per-key state store across batches. This is more complex but gives you fine-grained control over timeouts and eviction policies.

Common Mistakes

Mistake 1: Deduplicating without understanding the business key. If you drop rows without a clear policy on which duplicate to keep, you risk losing important data. Always ask: what makes a record unique in this domain? Order ID? User ID + timestamp? Always document this.

Mistake 2: Deduplicating after joins. Joins can multiply rows if the join keys aren't unique on both sides. Deduplicate before joining when possible, or validate that your join is 1:1. Deduplicating after a join may hide data quality issues.

Mistake 3: Ignoring late arrivals in streaming. If you deduuplicate on event_id in a stream without watermarks, early-arriving records might be deduplicated against late-arriving duplicates that arrive after the window is closed. Always set appropriate watermark delays.

Mistake 4: Using dropDuplicates with massive column lists. Specifying too many columns in dropDuplicates() is often a sign you don't fully understand your data model. Be precise about the true business key; extra columns just add noise and overhead.

Mistake 5: Not monitoring deduplication rates. If your deduplication suddenly removes 50% of records instead of 0.5%, something is wrong upstream. Log the before and after row counts, and alert on unusual changes.

Key Takeaways

• Use dropDuplicates() for simple, idempotent batch deduplication when all duplicates are identical.
• Use window functions with row_number() when you need to keep the latest, most complete, or highest-quality duplicate.
• Leverage source-system event IDs and anti-joins for the most robust, auditable deduplication strategy.
• In streaming, always set watermarks and understand the lateness guarantees before deduplicating; consider stateful operators like mapGroupsWithState() for complex scenarios.
• Document your deduplication policy per dataset: which columns form the business key, which duplicate row survives, and why.
• Monitor deduplication rates and row counts before/after to catch upstream issues early.
• Prefer deduplication as close to the source as possible; it's cheaper and more maintainable than downstream fixes.