Building Incremental and Idempotent Pipelines

Two properties, two different questions

Incremental processing updates a result from a bounded set of changes instead of rebuilding everything. It can still reread an overlap or recompute affected groups. A changed customer attribute may affect many historical rows, so the work is not always limited to the changed source row.

Idempotency means repeating the same logical operation has the same intended effect as applying it once. It does not prove that the result is correct or that processing happens exactly once. A full replacement can be idempotent without being incremental, and an incremental append can duplicate rows on retry. State the target and side effects covered by the guarantee.

The same date interval can produce different input after a correction. Bind a logical operation to retained input or a reproducible snapshot, a transformation version, and a target scope. A corrected delivery is new intent, not automatically a retry. An intentional logic change needs a new operation identity and a policy for replacing results produced by the old logic.

Fix the extraction boundary before running

An incremental extraction watermark records how far a source has been covered. Distinguish its clock or sequence from business event time, arrival time, and the scheduler’s interval. Choose fixed bounds such as (last_position, upper_position] under a source protocol that makes that range complete. A multi-partition source may need one position per partition rather than a single global number.

A timestamp plus a key handles ordering ties in a stable read; it does not discover a later commit carrying an older timestamp. A lookback is a bounded correction policy, not protection against arbitrarily late changes. Use stronger source change tracking or reconciliation when the bound cannot be justified. Current-row reads also need a way to discover hard deletes.

Persist the actual input reference, schema and transformation versions, extraction bounds, and validation evidence. A run that finds no changed rows may still advance a proven complete interval. No rows returned is not proof that the source was reachable or that extraction was complete. Missing input and a valid empty result must remain different states.

Choose what the destination operation means

For partition replacement, stage the complete desired contents of the declared partition, including an explicitly empty replacement. A change-only extract is not that complete snapshot. If an update moves a row between dates, repair both its old and new locations. Validate the candidate and publish it through the target’s supported atomic replacement or snapshot mechanism.

For upsert and merge, use stable non-null keys and an ordering version defined by the source. A delayed older version must not overwrite a newer one. Reject equal versions with different payloads instead of selecting whichever row Spark happened to encounter first. dropDuplicates on a key does not express a newest-version policy; use an explicit total order and reject unresolved ties.

Keep ordered delete evidence when an old replay could resurrect a removed key. The lab keeps a tombstone row with its version and excludes it from the live view. Its retention must cover permitted replays; removing both the row and deletion evidence can restore old data. Current-state upsert is not a complete history, and a guard against the latest version does not validate every past version’s consistency.

A transactional table’s MERGE must follow that engine’s duplicate-match, isolation, and conflict rules. A DataFrame transformation alone does not provide a transaction, and writing a plain Parquet directory is not a multi-file atomic commit. Separate candidate creation from publication; readers need a committed file set or table snapshot, not an in-progress directory listing.

Commit the effect and its evidence together

A batch load ledger binds a batch ID to immutable input and records successful application. If the ledger, target rows, and progress live in one transactional database, commit them together. A failure before commit should leave none of those changes visible. If commit succeeds but the response is lost, a retry can recognize the same committed operation.

When the destination and checkpoint are separate systems, simply saving the checkpoint after publication leaves a failure window. Make destination retries safe and reconcile the committed destination operation before advancing progress. Saving progress first can lose data. Multiple output tables or an external notification need their own supported publication protocol; one table’s commit does not make them atomic.

A fencing token identifies a writer generation that the target enforces. After an authorized replacement writer advances the generation, an old worker must be rejected at the mutation boundary. A scheduler lock or lease alone does not stop a paused worker from waking up and writing later. Generation issuance and validation must be serialized with the protected effect; merely adding a token field is insufficient.

The lab serializes SQLite writes using BEGIN IMMEDIATE and checks generation and source progress inside the transaction. It demonstrates the rule, not a distributed lock service. Real systems must handle lock contention, authorized generation issuance, failure recovery, and every write route. Restore compatible target, ledger, and progress state together; an old ledger against an empty target may skip necessary reconstruction.

Backfill deliberately and test interrupted runs

A backfill chooses historical bounds, retained inputs, and a transformation version. Isolate candidate output and coordinate publication with ordinary runs so an old backfill cannot overwrite a newer correction. Source ordering and transformation version are different dimensions: a newer entity version does not automatically express a new business rule for all history.

Check full result contents at small scale, and use reconciliation suited to larger data with explicit limits. Test repeated batches, changed content under the same ID, failure before commit, a lost success response, older updates after deletes, conflicting equal versions, empty intervals, skipped progress, and an obsolete writer. Counts and one checksum alone do not establish capture completeness.

For Spark foreachBatch, the callback can run again; design its destination effects to tolerate retries and scope batch identity to the query or checkpoint generation. Resetting a checkpoint can invalidate a scheme that treats a bare batch number as globally unique. The SQLite lab below tests transaction behavior with synthetic extracted batches. It does not test source CDC, Spark, Delta Lake, multiple writers, or machine-loss durability.

Lab: one transactional destination

Run the setup once with Python 3 and SQLite 3.24 or later, then each example. Each uses a fresh in-memory database. A row is (key, entity version, integer value, deleted flag); deleted rows use a null value. The source cursor is a synthetic complete-batch position, independent of entity versions. The code trusts the supplied extraction bounds; it cannot prove that no source rows are missing. Staging rejects duplicate key-version pairs. The digest binds normalized input, bounds, and the selected transform-v1 label, not an automatically detected code version. Query live rows with WHERE deleted=0. The examples retain tombstones and receipts without implementing cleanup.

import sqlite3, hashlib, json

def database():
    db = sqlite3.connect(":memory:", isolation_level=None)
    db.executescript("""
      CREATE TABLE state (id TEXT PRIMARY KEY NOT NULL, version INTEGER NOT NULL,
                          value INTEGER, deleted INTEGER NOT NULL);
      CREATE TABLE batches (id TEXT PRIMARY KEY NOT NULL, digest TEXT NOT NULL);
      CREATE TABLE progress (id INTEGER PRIMARY KEY, cursor INTEGER, epoch INTEGER);
      INSERT INTO progress VALUES (1, 0, 1);
    """)
    return db

def snapshot(db):
    return (db.execute("SELECT * FROM state ORDER BY id").fetchall(),
            db.execute("SELECT * FROM batches ORDER BY id").fetchall(),
            db.execute("SELECT cursor, epoch FROM progress").fetchone())

def apply(db, batch_id, start, end, rows, epoch, fail=False):
    if not batch_id or type(start) is not int or type(end) is not int or end <= start:
        raise ValueError("invalid batch bounds")
    seen = set()
    for key, version, value, deleted in rows:
        if not isinstance(key, str) or not key or type(version) is not int or version < 1:
            raise ValueError("invalid identity or version")
        if type(deleted) is not bool or (deleted and value is not None) or (not deleted and type(value) is not int):
            raise ValueError("invalid payload")
        if (key, version) in seen:
            raise ValueError("duplicate key-version in staged batch")
        seen.add((key, version))
    payload = ["transform-v1", start, end, sorted(rows, key=lambda r: (r[0], r[1]))]
    digest = hashlib.sha256(json.dumps(payload, separators=(",", ":")).encode()).hexdigest()
    db.execute("BEGIN IMMEDIATE")
    try:
        cursor, current_epoch = db.execute("SELECT cursor, epoch FROM progress WHERE id=1").fetchone()
        if epoch != current_epoch:
            raise ValueError("stale worker")
        receipt = db.execute("SELECT digest FROM batches WHERE id=?", (batch_id,)).fetchone()
        if receipt:
            if receipt[0] != digest:
                raise ValueError("batch identity conflict")
            db.commit()
            return "already committed"
        if start != cursor:
            raise ValueError("unexpected source cursor")
        for key, version, value, deleted in sorted(rows, key=lambda r: (r[0], r[1])):
            old = db.execute("SELECT version,value,deleted FROM state WHERE id=?", (key,)).fetchone()
            if old and version == old[0] and (value, int(deleted)) != old[1:]:
                raise ValueError("equal-version conflict")
            if old and version <= old[0]:
                continue
            db.execute("INSERT INTO state VALUES (?,?,?,?) ON CONFLICT(id) DO UPDATE SET version=excluded.version,value=excluded.value,deleted=excluded.deleted", (key,version,value,int(deleted)))
        if fail:
            raise RuntimeError("injected before ledger and cursor")
        db.execute("INSERT INTO batches VALUES (?,?)", (batch_id,digest))
        db.execute("UPDATE progress SET cursor=? WHERE id=1", (end,))
        db.commit()
        return "committed"
    except Exception:
        if db.in_transaction:
            db.rollback()
        raise

def rejects(action, error):
    try:
        action()
    except error:
        return True
    raise AssertionError("expected rejection")

Roll back an interrupted load, retry a committed batch, and reject changed input under the same ID.

db = database()
rows = [("A",1,10,False),("B",1,20,False)]
before = snapshot(db)
assert rejects(lambda: apply(db,"B1",0,1,rows,1,fail=True), RuntimeError)
assert snapshot(db) == before
assert apply(db,"B1",0,1,rows,1) == "committed"
committed = snapshot(db)
assert apply(db,"B1",0,1,rows,1) == "already committed"
assert snapshot(db) == committed
assert rejects(lambda: apply(db,"B1",0,1,[("A",1,99,False)],1), ValueError)
assert snapshot(db) == committed
print("failure rolled back state, ledger, and cursor:", True)
print("retry unchanged; reused ID with new input rejected:", True)
db.close()
# failure rolled back state, ledger, and cursor: True
# retry unchanged; reused ID with new input rejected: True

Apply a correction and deletion, then replay older versions and test a conflict and an empty interval.

db = database()
apply(db,"B1",0,1,[("A",1,10,False),("B",1,20,False)],1)
apply(db,"B2",1,2,[("A",2,12,False),("B",2,None,True)],1)
apply(db,"B3",2,3,[("A",1,10,False),("B",1,20,False)],1)
assert db.execute("SELECT * FROM state ORDER BY id").fetchall() == [("A",2,12,0),("B",2,None,1)]
before = snapshot(db)
assert rejects(lambda: apply(db,"B4",3,4,[("A",2,99,False)],1), ValueError)
assert snapshot(db) == before
assert apply(db,"empty",3,4,[],1) == "committed"
assert db.execute("SELECT cursor FROM progress").fetchone()[0] == 4
print("older replay cannot overwrite or resurrect:", True)
print("equal-version conflict rolled back; empty interval committed:", True)
db.close()
# older replay cannot overwrite or resurrect: True
# equal-version conflict rolled back; empty interval committed: True

Advance the authorized writer generation, reject the old worker, and reject a cursor gap.

db = database()
db.execute("BEGIN IMMEDIATE")
db.execute("UPDATE progress SET epoch=epoch+1 WHERE id=1")
db.commit()
before = snapshot(db)
assert rejects(lambda: apply(db,"old",0,1,[("A",1,10,False)],1), ValueError)
assert snapshot(db) == before
assert apply(db,"new",0,1,[("A",1,10,False)],2) == "committed"
assert rejects(lambda: apply(db,"gap",2,3,[],2), ValueError)
assert db.execute("SELECT cursor,epoch FROM progress").fetchone() == (1,2)
print("old worker rejected; current worker committed:", True)
print("noncontiguous cursor rejected:", True)
db.close()
# old worker rejected; current worker committed: True
# noncontiguous cursor rejected: True

References: SQLite transactions, Delta MERGE semantics, Spark Structured Streaming.