Idempotent Data Jobs

An operation is idempotent when applying it again with the same logical input leaves the same observable state. Data jobs need this property because schedulers retry, operators rerun failed partitions, and replayable sources can redeliver records. “The task ran once” is not a dependable control.

Prerequisites

Complete Pipeline Dependency Graphs. The source contract provides the stable event key event_id.

Learning goals

• Identify the logical operation key and partition.
• Replace append-only retry behavior with convergent publication.
• Test idempotency through published rows and totals.
• Distinguish idempotence from exactly-once execution.

Worked example

The fixture deliberately delivers e-101 in both files. The sink stores integer cents and enforces one row per event_id. ON CONFLICT ... DO UPDATE makes a replay converge on the latest representation instead of silently appending another logical event.

import json
import sqlite3
from decimal import Decimal
from pathlib import Path

root = Path("fixtures/data-engineering/pipelines")
connection = sqlite3.connect(":memory:")
connection.execute("""
    CREATE TABLE events (
        event_id TEXT PRIMARY KEY,
        occurred_at TEXT NOT NULL,
        arrived_at TEXT NOT NULL,
        customer_id TEXT NOT NULL,
        amount_cents INTEGER NOT NULL
    )
""")

def load(files):
    for name in files:
        for line in (root / name).read_text().splitlines():
            event = json.loads(line)
            cents = int(Decimal(event["amount"]) * 100)
            connection.execute("""
                INSERT INTO events VALUES (?, ?, ?, ?, ?)
                ON CONFLICT(event_id) DO UPDATE SET
                    occurred_at = excluded.occurred_at,
                    arrived_at = excluded.arrived_at,
                    customer_id = excluded.customer_id,
                    amount_cents = excluded.amount_cents
            """, (event["event_id"], event["occurred_at"], event["arrived_at"],
                  event["customer_id"], cents))
    connection.commit()

files = ["events-batch-1.jsonl", "events-batch-2.jsonl"]
counts = []
for _ in range(2):
    load(files)
    counts.append(connection.execute("SELECT COUNT(*) FROM events").fetchone()[0])
total = connection.execute("SELECT SUM(amount_cents) FROM events").fetchone()[0]
print(counts, total)

The observable result is [5, 5] 4250: processing six deliveries twice still produces five events worth 42.50. This is an idempotent sink effect. The Python statements still execute again, so this is not a claim that every line ran exactly once.

Useful publication patterns include an upsert keyed by a stable identity, replacing a complete partition in one transaction, and writing a versioned artifact before atomically changing a pointer or manifest. Choose conflict behavior deliberately: an event with the same key but different immutable fields may signal a source error and deserve quarantine rather than overwrite.

Limitations and common mistakes

• A generated key created during each attempt is not a stable logical identity.
• INSERT plus “check first” can race unless the database enforces uniqueness transactionally.
• Idempotent output does not make emails, payments, or other external side effects idempotent.
• Overwriting a partition in place can expose partial data; publish atomically.
• Deduplication needs a stated retention horizon; unbounded key state has unbounded cost.

Practice

Change e-101's amount in the second batch. Decide whether the contract permits correction, requires a new version, or quarantines the conflict. Then express that rule as a sink invariant.

Next steps

Continue with Incremental Processing to reduce scanned work without losing replay safety.

Primary sources

Checked 2026-10-15: Airflow best practices for repeatable tasks, SQLite UPSERT, and SQLite transactions.