Building Idempotent, Exactly-Once Batch Data Pipelines with Apache Spark

Modern data engineering teams face increasing pressure to deliver reliable, repeatable insights from massive batch data. But at scale, pipeline failures, retries, and duplicate processing can silently corrupt results—especially when exactly-once semantics are needed. In this post, I’ll break down how to build robust, idempotent batch pipelines with Apache Spark, ensuring data accuracy even under real-world failure modes.

What Are Exactly-Once and Idempotent Batch Pipelines?

Exactly-once batch processing guarantees each input record is processed a single time, even if the pipeline is retried due to failure. Idempotency ensures re-running a job with the same inputs produces the same outputs, without side effects or duplication. Both are critical for financial systems, analytics, and compliance workloads where data correctness is non-negotiable.

Defining Idempotency and Exactly-Once in Spark Pipelines

The challenge: Spark jobs often write to storage (S3, HDFS, Delta Lake, etc.) and downstream systems (databases, warehouses). A job failure or re-run can produce duplicate records or partial results. To combat this, we need to design jobs that:

• Detect and skip already-processed data
• Clean up partial outputs on retry
• Use atomic or transactional writes where possible

Here’s a simplified example using Delta Lake (version 2.4.0) with Spark 3.4.1, leveraging transactional writes and job run IDs for idempotency:

from pyspark.sql import SparkSession
from delta.tables import DeltaTable
import uuid

spark = SparkSession.builder \
    .appName('exactly-once-batch') \
    .config('spark.sql.extensions', 'io.delta.sql.DeltaSparkSessionExtension') \
    .config('spark.sql.catalog.spark_catalog', 'org.apache.spark.sql.delta.catalog.DeltaCatalog') \
    .getOrCreate()

run_id = str(uuid.uuid4())
input_df = spark.read.parquet('s3://my-bucket/input/')
input_df = input_df.withColumn('run_id', lit(run_id))
delta_path = 's3://my-bucket/output/'

delta_tbl = DeltaTable.forPath(spark, delta_path)
delta_tbl.alias('target').merge(
    input_df.alias('source'),
    'target.primary_key = source.primary_key AND target.run_id = source.run_id'
).whenNotMatchedInsertAll().execute()

Key insight: Using transactional sinks like Delta Lake with unique run IDs lets Spark batch jobs be retried safely, ensuring exactly-once semantics.

Step 1: Choose Idempotent File Formats and Sinks

Why Output Format and Sink Choice Matters

The choice of output format and sink determines whether you can guarantee atomic writes, support upserts, or clean up failed job outputs. In practice, I recommend these options (with supporting versions as of 2024):

• Delta Lake (v2.4.0+): ACID transactions, upserts (MERGE), schema evolution.
• Apache Iceberg (v1.3.0+): Hidden partitioning, snapshot isolation, rollback.
• BigQuery (with WRITE_TRUNCATE or MERGE DML): Atomic batch loads in the cloud.

Avoid plain CSV/Parquet on S3/GCS/HDFS for production idempotency—these formats lack atomicity. Using transactional data lakes or warehouse tables is essential.

Implementation Notes

• For S3 or GCS, ensure table versioning and lock management is enabled (Delta: S3DynamoDB lock; Iceberg: AWS Glue or Hive).
• For warehouse sinks, use atomic batch operations (e.g., BigQuery’s MERGE statement or Snowflake’s multi-table transactions).

Key insight: Transactional table formats (Delta, Iceberg) and ACID-compliant warehouses are the foundation for robust, idempotent batch pipelines.

Step 2: Implement Unique Run Identifiers for Each Pipeline Execution

Assigning and Propagating Run IDs

A unique run ID disambiguates outputs from multiple pipeline runs, ensuring that reprocessing or retries don’t create duplicates. In Spark, I inject a UUID at the start of each job and propagate it as a column throughout the pipeline.

Why This Works

With a run_id column, downstream tables or consumers can identify and deduplicate records if needed. Combined with transactional sinks, this enables safe upserts (idempotent by design) and easier debugging.

Example: Adding Run ID in PySpark

from pyspark.sql.functions import lit
run_id = str(uuid.uuid4())
df_with_run_id = df.withColumn('run_id', lit(run_id))

Best Practices

• Store the run_id in pipeline metadata (e.g., Airflow XCom, Databricks job tags).
• Add run_id as a partition column in your output table for fast filtering and troubleshooting.
• For daily or hourly batch jobs, use a deterministic run ID (e.g., YYYYMMDDHH) for idempotency across reruns.

Key insight: Explicit run IDs make batch outputs traceable, auditable, and safe to deduplicate during recovery.

Step 3: Use Atomic Upserts or MERGE Operations for Output Writes

Why MERGE Beats Overwrite for Idempotency

In Spark, plain overwrite mode can leave partial results if the job fails mid-write. Instead, use atomic MERGE or upsert operations supported by Delta Lake, Iceberg, and major warehouses. This guarantees only new or changed records are written, and retries do not duplicate data.

Example: Delta Lake MERGE for Idempotent Upserts

deltaTable = DeltaTable.forPath(spark, delta_path)
deltaTable.alias('target').merge(
    source=staged_df.alias('source'),
    condition='target.id = source.id'
).whenMatchedUpdateAll()
 .whenNotMatchedInsertAll()
 .execute()

• In Iceberg (v1.3.0+), use MERGE INTO SQL syntax.
• In BigQuery, use MERGE DML statements.

Handling Partial Writes and Recovery

If your pipeline crashes, atomic upserts ensure that only the intended changes are committed. Uncommitted transactions are rolled back, so a retry simply re-applies the batch safely.

Key insight: MERGE/upsert operations are essential for exactly-once guarantees—never rely on overwrite mode for production batch jobs.

Step 4: Enforce Idempotency Upstream with Deduplication and Watermarking

Preventing Duplicate Input Records

Even with idempotent outputs, duplicate source data or late-arriving events can still cause problems. To guarantee correctness, implement deduplication and watermarking upstream:

1. Deduplicate by Primary Key

If your source data has a unique key (e.g., order_id), use Spark’s dropDuplicates or SQL ROW_NUMBER() windowing to select only the latest record per key.

deduped_df = df.dropDuplicates(['primary_key'])

2. Watermarking for Bounded Input

For append-only data, process only new records since the last successful run, tracked via max-timestamp or offset. Store this watermark in a metadata store (e.g., S3, DynamoDB, PostgreSQL).

3. Integrate with Orchestration Tools

Use orchestration frameworks (Airflow 2.8, Dagster, or Databricks Workflows) to track the status of each batch job, commit watermarks, and trigger retries as needed.

Handling Late Data and Reprocessing

In compliance workloads, you may need to reprocess late-arriving data. By combining watermarks with run IDs, you can safely process only the missing ranges without affecting previous results.

Key insight: Upstream deduplication and watermarking are critical for end-to-end idempotency—not just at the output stage.

Step 5: Automate Cleanup and Recovery Logic in Orchestration

Why Manual Cleanup Fails at Scale

At petabyte scale, a failed batch job can leave orphaned files, partial partitions, or duplicate outputs. Manual intervention is slow and error-prone. Instead, automate cleanup and recovery using orchestration frameworks:

Orchestration Patterns

1. Pre-Job Cleanup: Before each run, delete or archive unfinished output directories/partitions based on run ID or timestamp.
2. Post-Failure Rollback: Use Delta Lake’s VACUUM and time travel features to roll back to the last consistent state. For Iceberg, use snapshot rollback APIs.
3. Automated Retry Policy: Configure Airflow’s max_retries and retry_delay, or use Databricks job clusters with automated retry semantics.

Example: Airflow Task for Cleanup

from airflow.operators.python_operator import PythonOperator

def cleanup_partial_outputs(**context):
    run_id = context['ti'].xcom_pull(task_ids='generate_run_id')
    # Delete S3 folder with this run ID
    # ... S3 cleanup code ...

cleanup_task = PythonOperator(
    task_id='cleanup_output',
    python_callable=cleanup_partial_outputs,
    provide_context=True
)

Key insight: Automated cleanup and rollback logic in your orchestrator is indispensable for production-ready, self-healing data pipelines.

Comparison Table: Delta Lake vs. Iceberg vs. BigQuery for Idempotent Batch Pipelines

Feature	Delta Lake (v2.4.0+)	Iceberg (v1.3.0+)	BigQuery
ACID Transactions	Yes	Yes	Yes
Native Upsert/MERGE	Yes	Yes	Yes
Schema Evolution	Yes	Yes	Yes
Partition Evolution	Limited	Yes	No
Time Travel	Yes	Yes	Yes
Rollback Support	Yes	Yes	Limited
S3/GCS Direct Support	Yes	Yes	No
Orchestration Integration	Excellent (Databricks, Airflow)	Good (Airflow, Dagster)	Excellent (Cloud Composer, Airflow)

Key insight: Delta Lake and Iceberg excel for idempotent batch pipelines on cloud object storage, while BigQuery is optimal for cloud-native warehouses.

Frequently Asked Questions

Q: What’s the difference between idempotency and exactly-once in data pipelines? A: Idempotency ensures rerunning a pipeline with the same input does not create duplicate or inconsistent outputs. Exactly-once semantics guarantee each input is processed only once, even with retries. In batch data engineering, both work together to prevent duplication and data corruption.

Q: How do I ensure a Spark batch job is safe to retry after failure? A: Use transactional sinks (Delta Lake, Iceberg, or BigQuery), assign a unique run ID to each execution, and write outputs via upsert/MERGE instead of overwrite. Automate cleanup and watermarking in your orchestrator to avoid partial results.

Q: Can I achieve exactly-once processing with plain Parquet or CSV files on S3? A: No. Plain Parquet or CSV files on object storage do not support atomic transactions or upserts, making them unsafe for exactly-once requirements. Use Delta Lake or Iceberg table formats for strong guarantees on S3 or GCS.

Key Takeaways

• Use ACID-compliant table formats (Delta Lake, Iceberg) or warehouses (BigQuery) for idempotent, exactly-once batch pipelines.
• Assign unique run IDs to each pipeline execution and propagate them with your data for traceability and safe retries.
• Prefer upsert/MERGE operations over overwrite for writing outputs in Spark, ensuring atomicity and deduplication.
• Implement deduplication and watermarking upstream to prevent duplicate or late-arriving input records from corrupting results.
• Automate cleanup, rollback, and recovery logic in orchestrators (Airflow, Databricks Workflows) for self-healing pipelines.
• Never rely on plain file outputs (Parquet, CSV) on S3/GCS for production-grade idempotency or exactly-once guarantees.