Handling Concurrent Data Loads in Delta Tables

Ensuring Reliable Concurrent Writes With Retrying Options

Delta Lake is a resilient storage layer that offers ACID transactions, schema enforcement, and data versioning. However, concurrent writes generate contention since different processes are attempting to write, update, or delete at the same time. This process offers a structured retry mechanism with exponential backoff to handle concurrency in Delta tables.

Delta Table Concurrent Writes Issues

Concurrency failures occur when multiple processes simultaneously attempt to write to the same Delta table. The common failure scenarios are as follows:

• ConcurrentAppendException – When concurrent jobs append records simultaneously, with conflicting appends.
• ConcurrentDeleteReadException – If a process is trying to read deleted data by another process.
• ConcurrentDeleteDeleteException – When two processes attempt to delete the same data simultaneously.

These issues must have an ever-present retry facility that makes writes happen successfully with consistency.

Proposed Retry Mechanism for Delta Table Writes

A streaming write retry procedure is employed to mitigate concurrent write failures utilizing exponential backoff.

The following Python code describes the process:

from datetime import datetime
from time import sleep
from delta.exceptions import (
    ConcurrentAppendException,
    ConcurrentDeleteReadException,
    ConcurrentDeleteDeleteException,
)
import math

def streaming_write_with_concurrent_retry(
    stream, max_attempts=3, indefinite=False, table=None, path=None
):
    """
    Handles concurrent write operations to a Delta table or path by retrying the operation
    in case of specific concurrent exceptions.

    :param stream: The data stream to be written.
    :param max_attempts: The maximum number of retry attempts. Default is 3.
    :param indefinite: If True, will keep retrying indefinitely. Default is False.
    :param table: The Delta table to write to.
    :param path: The path to write to.
    :return: The result of writer.awaitTermination().
    """

    attempt = 0  # Initialize attempt counter

    while True:
        try:
            # Choose the writer based on whether table or path is provided
            if table:
                writer = stream.table(table)
            elif path:
                writer = stream.start(path)
            else:
                writer = stream.start()

            # Attempt to write and wait for termination
            return writer.awaitTermination()

        # Handle concurrent exceptions
        except (
            ConcurrentAppendException,
            ConcurrentDeleteReadException,
            ConcurrentDeleteDeleteException,
        ) as e:

            # Increment attempt counter
            attempt += 1

            # If indefinite is False and attempts have reached max_attempts, raise the exception
            if not indefinite and attempt >= max_attempts:
                raise e from None

            # Calculate sleep time using exponential backoff strategy
            sleep_time = min(120, math.pow(2, attempt))

            # Log the retry attempt
            print(f"Retrying {attempt}/{max_attempts if not indefinite else '∞'} after {sleep_time} seconds due to {type(e).__name__}")

            # Sleep for the calculated time before retrying
            sleep(sleep_time)

Retry Strategy Explanation

The retry policy follows an exponential backoff policy:

1. Exception Identification

The function catches ConcurrentAppendException, ConcurrentDeleteReadException, and ConcurrentDeleteDeleteException.

2. Retry Attempts and Limitations

The function retries a maximum of max_attempts times before it fails. The indefinite=True parameter allows infinite retries until success.

3. Exponential Backoff Calculation

Backoff formula: sleep_time = min(120, 2^attempt). This ensures retry wait times grow exponentially but are capped at a maximum of 120 seconds.

Example of retry wait times:

• Attempt 1 → Wait 2 seconds
• Attempt 2 → Wait 4 seconds
• Attempt 3 → Wait 8 seconds
• (up to a maximum of 120 seconds)

Resuming the Stream

Once the retry is successful, writer.awaitTermination() enables the streaming job to continue running.

Alternative Strategies for Delta Table Concurrency Handling

Besides retry-based conflict resolution, Delta Lake offers additional techniques:

1. Optimistic Concurrency Control (OCC)

Delta Lake checks for conflicts before committing a transaction. If there is a conflict, it will retry the operation automatically.

2. Partitioning Data for Isolation

Write operations should be targeted at other partitions to avoid collision.

df.write.format("delta").mode("overwrite").option("replaceWhere", "date = '2025-02-17'").save("/mnt/delta/table")

This limits updates to a single partition, reducing contention.

3. Streaming and Auto-Optimize 

Enable Auto-Optimize and Auto-Compact:

ALTER TABLE delta.`/mnt/delta/table_name` SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true'
)

This config change reduces small files and improves concurrent write performance.

4. Merge-Based Upserts

Instead of direct inserts, use MERGE to handle conflicts:

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/mnt/delta/table_name")

delta_table.alias("target").merge(
    df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(set={"target.value": "source.value"}).whenNotMatchedInsertAll().execute()

This process ensures row-level conflict resolution.

Monitoring Debugging Concurrency Issues

1. Check data load transaction history:

DESCRIBE HISTORY delta.`/mnt/delta/table_name`;

2. View/check any active locks:

SHOW TBLPROPERTIES delta.`/mnt/delta/table_name`;

3. Enable the change data feed (CDF) for change tracking:

ALTER TABLE delta.`/mnt/delta/table_name` SET TBLPROPERTIES ('delta.enableChangeDataFeed' = true);

Conclusion

Delta tables allow concurrent writes with ACID guarantees, but conflicts are possible.

• A retry-based strategy with exponential backoff helps mitigate concurrency issues.
• Partitioning, MERGE, and Auto-Optimize also improve concurrent write performance.
• Tracking mechanisms such as DESCRIBE HISTORY and CDF follow conflicts.

By adhering to these best practices, the process can efficiently attain concurrent delta table writes, maintain data integrity, and achieve performance optimizations.