Rana Nasri Ghazzi

Handling Duplicates in Spark Streaming Pipelines:

3 Practical Approaches

Duplicate records are one of the most common and frustrating problems in data engineering. Whether you’re pulling from an API, a message queue, or a CDC feed, duplicates will eventually appear. Here’s a breakdown of three practical approaches to deal with them — and when to use each.

The Problem

In streaming or periodic ingestion pipelines, duplicates can sneak in through:

• Retried API calls after a timeout
• At-least-once delivery guarantees in message brokers like Kafka
• Overlapping watermark windows during micro-batch processing
• Pipeline restarts replaying already-processed records

Left unchecked, duplicates silently corrupt aggregations, inflate metrics, and break downstream consumers.

Option 1: dropDuplicates() Before Writing

The simplest fix — deduplicate the DataFrame before it hits the sink.

def process_batch(df, epoch_id):
    df_deduped = df.dropDuplicates(["id", "event_timestamp"])
    df_deduped.write \
        .format("delta") \
        .mode("append") \
        .save("/mnt/delta/events")

How it works: Spark compares rows within a single micro-batch and drops exact duplicates based on the columns you specify.

Pros:

• Simple, one-liner fix
• No external state needed
• Works well when duplicates appear within the same batch

Cons:

• Does not catch cross-batch duplicates — if the same record arrives in two different micro-batches, it still gets written twice
• On large streams, deduplication across a wide key space can be memory-intensive

Use when: Duplicates are caused by source-side fan-out within a single batch and cross-batch duplication is not a concern.

Option 2: MERGE with foreachBatch (Recommended for Streaming)

Instead of appending blindly, use a MERGE (upsert) statement inside foreachBatch. This compares incoming records against what already exists in the Delta table and only inserts records that aren’t already there.

from delta.tables import DeltaTable

def upsert_to_delta(micro_batch_df, epoch_id):
    target = DeltaTable.forPath(spark, "/mnt/delta/events")
    
    target.alias("target").merge(
        micro_batch_df.alias("source"),
        "target.id = source.id AND target.event_timestamp = source.event_timestamp"
    ) \
    .whenNotMatchedInsertAll() \
    .execute()

stream_query = (
    df_stream
    .writeStream
    .foreachBatch(upsert_to_delta)
    .outputMode("update")
    .start()
)

How it works: For each micro-batch, Delta Lake checks if the incoming row already exists in the target table. If it does, it skips it (or updates it, your choice). If it doesn’t, it inserts it.

Pros:

• Catches duplicates across batches — the most robust streaming option
• Idempotent: re-running the same batch won’t create duplicates
• Leverages Delta Lake’s ACID guarantees

Cons:

• Slightly more complex to set up
• MERGE on large tables can be expensive without proper partitioning and Z-ordering on the merge keys

Use when: You’re running a true streaming pipeline with continuous micro-batches and need cross-batch duplicate protection.

Option 3: Batch MERGE (Most Appropriate for Periodic API Pulls)

If your data source is a periodic API call — say, every 15 minutes or once an hour — you’re not really import logging
from delta.tables import DeltaTable

def load_to_delta(df_new, table_path, merge_keys):
# Merge new data into delta table using merge keys
target = DeltaTable.forPath(spark, table_path)

import logging
from delta.tables import DeltaTable

def  load_to_delta(df_new, table_path, merge_keys):
    # Merge new data into delta table using merge keys
    
target = DeltaTable.forPath(spark, table_path)
    
    merge_condition = " AND ".join([f"target.{k} = source.{k}" for k in merge_keys])
    
    (
        target.alias("target")
        .merge(df_new.alias("source"), merge_condition)
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )
    
    print(f"[load] merge complete on keys: {merge_keys}")
    logging.info(f"MERGE complete — keys: {merge_keys}")

Usage:

df_new = extract_from_api()          # pull latest batch
df_new = transform(df_new)           # clean, cast, normalize
load_to_delta(df_new, "/mnt/delta/events", merge_keys=["id"])

Pros:

• No stream state, no watermarks
• Fully idempotent: running the same pull twice produces the same result
• Easier to test, debug, and backfill
• Your watermark filter already eliminates most duplicates; MERGE is just the safety net

Cons:

• Not suitable for low-latency, sub-second ingestion
• Requires Delta Lake (or similar ACID-capable format)

Use when: Your pipeline is driven by a scheduler (Airflow, cron, Databricks Jobs) and pulls from an API, flat file, or REST endpoint on a schedule.

Comparison at a Glance

Key Takeaway

Match your deduplication strategy to your ingestion pattern:

• Continuous stream → foreachBatch with MERGE
• Scheduled batch pull → batch MERGE (simplest, most reliable)
• Quick in-batch cleanup → dropDuplicates() as a first pass

The watermark filter handles the bulk of stale data. MERGE is the idempotency guarantee that makes your pipeline safe to retry, backfill, or re-run without consequence — and in production pipelines, that safety net is worth having.