Configuring Idempotent Writes with Delta Lake MERGE in foreachBatch

Achieving Idempotency in foreachBatch

To guarantee exactly-once semantics when using foreachBatch with Delta Lake, the MERGE operation must rely on a deterministic match condition based on a unique business key or a composite key present in the data. A deterministic condition is one that produces the same result for the same input across different execution attempts, regardless of when or where the task is retried.

Deterministic vs. Non-Deterministic Conditions

A match condition is considered deterministic if it uses stable column references or deterministic functions. Examples include:

• Simple Column References: target.userId = source.userId
• Deterministic Functions: coalesce(source.id, 'unknown') or lower(source.email)
• Constants: Matching against a fixed value for specific logic.

Conversely, incorporating non-deterministic functions—such as current_timestamp(), rand(), or uuid()—inside the ON clause of a MERGE statement invalidates the idempotency guarantee. Because these functions evaluate differently during a retry, a record that was previously matched and updated might be treated as a new record and inserted again, leading to duplicates.

How Spark and Delta Lake Handle Retries

Spark Structured Streaming ensures that the foreachBatch function is executed for a specific set of offsets. However, if the executor fails after the data is written to the Delta table but before the offset is committed to the checkpoint, Spark will retry the entire batch.

Delta Lake manages this through atomic transactions. The MERGE operation is an atomic commit in the Delta Log. If the match condition is deterministic, the retry attempt will find the rows inserted by the failed attempt and update them (upsert) rather than appending them. If the match condition is non-deterministic, the retry will fail to find the previous records, resulting in duplicate entries.

Implementation Steps for Idempotent MERGE

1. Identify a Unique Key: Ensure your source data has a unique identifier (e.g., transaction_id).
2. Define a Stable Match Condition: Use only column-to-column comparisons or deterministic transformations in the ON clause.
3. Execute MERGE: Use the DeltaTable.merge() API within the foreachBatch block.

def upsertToDelta(df, batchId):
    # Use a deterministic match condition
    deltaTable = DeltaTable.forPath(spark, "/path/to/table")
    deltaTable.alias("target") \ 
        .merge(df.alias("source"), "target.id = source.id") \ 
        .whenMatchedUpdateAll() \ 
        .whenNotMatchedInsertAll() \ 
        .execute()

Verification

To verify idempotency, you can manually kill the Spark driver or executor during the write phase of a micro-batch. After restarting, run DESCRIBE HISTORY table_name to ensure the MERGE operation was recorded as a single atomic commit and check for duplicate IDs using a COUNT and GROUP BY on your primary key.

Diagnostic Detail Needed: Are you using a generated surrogate key (like a UUID) created at runtime, or a natural key provided by the source system?