Transactional Writer: All-or-Nothing Guarantees Against Partial Data

Series: Data Engineering Design Patterns | Post 12 of 64

You’re saving 60% on infrastructure by running batch jobs on spot instances. Great trade-off — until your downstream consumers start complaining about data quality. Half-written files. Duplicate records. Partial datasets that look complete but aren’t.

The issue: when the cloud provider reclaims a spot instance mid-job, tasks fail and retry on different nodes. But by then, some output has already been written. Now consumers see a mix of old, new, and partially-new data.

The Transactional Writer pattern solves this with database-level transactions: changes are invisible to readers until you explicitly commit them.

The Problem

Your batch job processes device events and writes results to a target table. The execution looks like this:

Spot instance starts → Task writes 70% of data → Spot instance terminated
                                ↓
                       Task retries on new node
                                ↓
                       Task writes 100% of data
                                ↓
                       Result: 170% of expected data
                       (70% from interrupted task + 100% from retry)

Without transactions, every partial write is visible. Your “complete” output is actually corrupted.

You need a guarantee: all the records, or none of them. Never something in between.

The Solution: Transactional Writer

Transactions provide all-or-nothing semantics. Three steps:

1. BEGIN                2. WRITE                3. COMMIT
   ─────────               ────────                ─────────
   ┌──────────┐           ┌──────────┐            ┌──────────┐
   │ Producer │           │ Producer │            │ Producer │
   └────┬─────┘           └────┬─────┘            └────┬─────┘
        │ START                │ INSERT/UPDATE         │ COMMIT
        ▼                      ▼                       ▼
   ┌─────────────────────────────────────────────────────────┐
   │                      Database                            │
   │  ┌─────────────────┐   ┌─────────────────┐              │
   │  │ Existing data   │   │ In-progress     │              │
   │  │ (visible)       │   │ (private to tx) │              │
   │  └────────┬────────┘   └────────┬────────┘              │
   └───────────┼─────────────────────┼───────────────────────┘
               │                     │
               ▼                     ▼
        ┌──────────┐           ┌──────────┐
        │ Consumer │           │ Consumer │
        └──────────┘           └──────────┘
        (sees old data)      (sees old data
                             until commit)

   After COMMIT: all changes visible at once
   On failure: ROLLBACK discards everything

The magic: consumers either see the complete new state or the previous complete state. Never an intermediate, partial state.

Implementation

SQL Transactions

The simplest form — using BEGIN, COMMIT, and ROLLBACK explicitly:

BEGIN;
-- Load first dataset
CREATE TEMPORARY TABLE changed_devices_file1 (LIKE devices);
COPY changed_devices_file1 FROM '/data_to_load/dataset_1.csv'
  CSV DELIMITER ';' HEADER;
MERGE INTO devices AS d
USING changed_devices_file1 AS c_d
ON d.device_id = c_d.device_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...);
-- Load second dataset
CREATE TEMPORARY TABLE changed_devices_file2 (LIKE devices);
COPY changed_devices_file2 FROM '/data_to_load/dataset_too_long_type.csv'
  CSV DELIMITER ';' HEADER;
MERGE INTO devices AS d
USING changed_devices_file2 AS c_d
ON d.device_id = c_d.device_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT (...);
COMMIT;

If the second file has bad data (e.g., values too long for some columns), the second MERGE fails. The first MERGE doesn’t commit either. The transaction rolls back automatically.

Result: Consumers never see the partial state where only file1’s changes were applied.

Delta Lake (Transactional File Format)

Delta Lake provides transactions on object stores. The transaction log makes writes invisible until committed:

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

spark = SparkSession.builder \
    .appName("TransactionalWriter") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", 
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()
# Read source data
processed_data = (
    spark.read.format("parquet")
    .load("/staging/devices")
    .transform(apply_business_logic)
)
# Write transactionally - Delta Lake handles atomicity
(
    processed_data.write
    .format("delta")
    .mode("overwrite")
    .save("/data/devices")
)
# Behind the scenes:
# 1. Spark writes data files (invisible to readers)
# 2. Once all files are written, Delta creates a commit log entry
# 3. Only then are the files considered "current"
# 4. If anything fails, no commit log entry → files orphaned but invisible

The key insight: Delta Lake files become visible only when the corresponding commit log file is created. Partial writes leave orphan data files that no reader will ever see.

Apache Flink with Kafka Transactions

For streaming, Kafka transactional producers via Flink:

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import (
    KafkaSink, 
    KafkaRecordSerializationSchema, 
    DeliveryGuarantee
)
from pyflink.common.serialization import SimpleStringSchema

env = StreamExecutionEnvironment.get_execution_environment()
# Configure transactional Kafka producer
kafka_sink = (
    KafkaSink.builder()
    .set_bootstrap_servers("localhost:9094")
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
        .set_topic("reduced_visits")
        .set_value_serialization_schema(SimpleStringSchema())
        .build()
    )
    .set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE)  # ← The magic
    .set_property("transaction.timeout.ms", str(1 * 60 * 1000))  # 1 minute
    .build()
)
# Use the transactional sink
stream.sink_to(kafka_sink)
env.execute("Transactional Stream Job")

Two critical settings:

delivery_guarantee = EXACTLY_ONCE: Enables Kafka transactions. Each Flink checkpoint commits a transaction containing all messages produced since the last checkpoint.

transaction.timeout.ms: How long Kafka waits before forcibly aborting an open transaction. Must be longer than your checkpoint interval, or transactions will expire before they can commit.

Two Implementation Strategies

For distributed processing jobs (Spark, Flink), there are two ways to use transactions:

Local transactions (per-task):

def write_partition_transactional(partition_data):
    """Each task opens its own transaction."""
    conn = get_db_connection()
    try:
        conn.begin()
        for row in partition_data:
            conn.execute("INSERT INTO target VALUES (...)", row)
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
df.foreachPartition(write_partition_transactional)

Simple to implement. But if the job retries after some tasks already committed, those records get inserted again. Idempotency is per-task, not per-job.

Job-level transactions (whole job atomic):

# Delta Lake handles this automatically
# Tasks write files privately, single commit at the end
df.write.format("delta").mode("overwrite").save("/data/target")

Stronger guarantee — the entire job is atomic. But harder to implement (requires coordination across tasks).

⚠️ Gotchas and Trade-offs

1. Commit Step Adds Latency

Non-transactional writes are visible immediately. Transactional writes have to wait for the commit:

JSON/CSV files:        Immediately visible after each task
Delta Lake files:      Visible only after the commit log is written

Consumers wait for the slowest task. This is the cost of atomicity. Usually worth it, but if you need sub-second visibility, transactions may not fit.

2. Distributed Framework Support Is Uneven

Not every combination works. For example:

Apache Spark + Delta Lake: Excellent transactional support

Apache Spark + Kafka transactions: Not supported natively (Spark can’t act as a transactional Kafka producer)

Apache Flink + Kafka: Excellent support via two-phase commit

Apache Flink + Delta Lake: Works but less mature than Spark

Check your combination carefully. The pattern only works if your specific stack supports it.

3. Idempotency Is Transaction-Scoped Only

A transactional writer prevents partial reads within a single job execution. It does not prevent duplicates across multiple executions.

Run 1: BEGIN → INSERT 1000 rows → COMMIT  → Table has 1000 rows
Run 2: BEGIN → INSERT 1000 rows → COMMIT  → Table has 2000 rows (duplicates!)

For cross-run idempotency, combine with patterns like Keyed Idempotency, Data Overwrite, or Merger.

4. READ UNCOMMITTED Defeats the Pattern

Some database clients can be configured to read uncommitted data:

-- Postgres
SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;

With READ UNCOMMITTED, consumers can see your in-progress transaction’s changes — including changes you eventually rollback. This is called a dirty read.

Make sure downstream consumers use at least READ COMMITTED isolation level. Most defaults do, but it’s worth verifying.

5. Long Transactions Hold Locks

While your transaction is open, the database holds locks on affected rows or files. Long-running transactions can:

• Block other writers
• Bloat database storage (uncommitted changes accumulate)
• Eventually time out and rollback all your work

For large batch jobs, consider breaking work into smaller transactional chunks:

BATCH_SIZE = 100_000

def process_in_chunks(data):
    for i in range(0, len(data), BATCH_SIZE):
        chunk = data[i:i + BATCH_SIZE]
        process_transactionally(chunk)  # Smaller, faster transaction

6. Transaction Timeout Matters

For Kafka transactions, the timeout must be tuned carefully:

# Too short: transactions expire before commit
.set_property("transaction.timeout.ms", "10000")  # 10 seconds

# Too long: failed jobs hold locks too long
.set_property("transaction.timeout.ms", "3600000")  # 1 hour

Rule of thumb: set it to 2–3x your expected checkpoint interval.

When to Use Transactional Writer

✅ Use it when:

• Running on unreliable infrastructure (spot instances, preemptible VMs)
• Consumers cannot tolerate partial data
• Multiple writes must succeed or fail together
• Working with transactional storage (Delta Lake, Iceberg, RDBMS, Kafka)

❌ Avoid it when:

• Your storage doesn’t support transactions (plain Parquet on S3)
• You need sub-second consumer visibility
• The framework + storage combination isn’t supported
• Cross-run idempotency is the actual requirement (use other patterns)

Combining with Other Patterns

Transactional Writer rarely works alone. Common combinations:

Transactional Writer + Keyed Idempotency: Atomic writes with cross-run idempotency

Transactional Writer + Data Overwrite: Atomic overwrites of partitions

Transactional Writer + Merger: Atomic UPSERT operations

Transactional Writer + Checkpointer: Streaming exactly-once delivery

Related Patterns

The Transactional Writer pattern works with:

• Keyed Idempotency: Use together for full exactly-once semantics
• Merger: Wrap MERGE operations in transactions for atomicity
• Checkpointer: Coordinate transactions with streaming checkpoints
• Proxy: Combine for immutable, transactionally-published datasets

Next Post

We’re closing the Idempotency chapter! In the next article, we’ll start the Data Value section with the Proxy Pattern — how to expose immutable datasets through a single access point while keeping historical versions intact.

Have you been bitten by partial writes from spot instance failures? How did you solve it? Share your experience in the comments.

I’m Franco, building AI systems at Convolution AI. If you’re working on AI or Data, let’s connect.

This post is part of a 64-article series covering all patterns from “Data Engineering Design Patterns” by Bartosz Konieczny (O’Reilly, 2025).