Incremental Processing

Incremental processing limits work to data that is new or changed relative to durable state. It can reduce latency and cost, but it adds state that must be correct, recoverable, and versioned with the transformation. A high-water mark is an optimization, not proof that no record was missed.

Prerequisites

Complete Idempotent Data Jobs. Incremental reads depend on an idempotent sink because overlap and replay are safety tools.

Learning goals

• Choose a checkpoint field whose ordering and meaning are documented.
• Advance a checkpoint only after successful publication.
• Combine overlap or replay with key-based deduplication.
• Rebuild when transformation logic or source history changes.

Worked example

This bounded example checkpoints source arrived_at, not business occurred_at. It writes the output and checkpoint in one SQLite transaction. Replaying batch 2 after the checkpoint advances selects no deliveries; the sink still protects against its duplicated e-101.

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

root = Path("fixtures/data-engineering/pipelines")
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE events(event_id TEXT PRIMARY KEY, amount_cents INTEGER NOT NULL)")
db.execute("CREATE TABLE state(source TEXT PRIMARY KEY, watermark TEXT NOT NULL)")

def process(name):
    watermark_row = db.execute(
        "SELECT watermark FROM state WHERE source = 'events'"
    ).fetchone()
    watermark = watermark_row[0] if watermark_row else ""
    deliveries = [json.loads(line) for line in (root / name).read_text().splitlines()]
    selected = [event for event in deliveries if event["arrived_at"] > watermark]
    with db:
        for event in selected:
            cents = int(Decimal(event["amount"]) * 100)
            db.execute(
                "INSERT INTO events VALUES (?, ?) "
                "ON CONFLICT(event_id) DO UPDATE SET amount_cents = excluded.amount_cents",
                (event["event_id"], cents),
            )
        if selected:
            db.execute(
                "INSERT INTO state VALUES ('events', ?) "
                "ON CONFLICT(source) DO UPDATE SET watermark = excluded.watermark",
                (max(event["arrived_at"] for event in selected),),
            )
    return len(selected)

processed = [process("events-batch-1.jsonl"), process("events-batch-2.jsonl"),
             process("events-batch-2.jsonl")]
unique_events = db.execute("SELECT COUNT(*) FROM events").fetchone()[0]
print(processed, unique_events)

The result is [3, 3, 0] 5: three deliveries in each new file, then none on the immediate replay, with five unique events. In a real source, timestamps can tie, arrive out of order, or change. Prefer a monotonic source offset or a composite cursor such as (updated_at, primary_key). If that is unavailable, reread a bounded overlap and rely on idempotent publication.

Keep a full-rebuild path. Checkpoints cannot repair results produced by old transformation code, deleted source rows, or source corrections outside the lookback. Reconcile incremental output against periodic source counts, sums, and key coverage.

Limitations and common mistakes

• MAX(event_time) misses records that arrive later with older event time.
• Advancing state before output commits can create permanent gaps.
• Advancing output without state atomically can repeat work; that is safe only with an idempotent sink.
• Mutable sources need update and deletion semantics, not only an append cursor.
• A checkpoint without transformation version and source identity is ambiguous.

Practice

Design a composite checkpoint for a source with second-resolution updated_at values and many records per second. State how deletes are represented and how a full rebuild is triggered.

Next steps

Continue with Late-Arriving Data and Watermarks, which separates arrival progress from event-time completeness.

Primary sources

Checked 2026-10-15: Airflow best practices on partition-specific I/O, SQLite transactions, and Apache Flink checkpoints.