Batch pipeline design

ETL, ELT and where transformation lives

ETL transforms data in flight, between source and destination, in a processing engine (Spark, a Python job, a dedicated tool), and loads the finished shape. ELT loads raw data into the warehouse or lakehouse first and transforms it there, in SQL, using the destination's compute. Cheap, elastic warehouse compute and tools like dbt made ELT the default for analytics: raw data is preserved, transformations are versioned SQL, and the warehouse's optimiser does the heavy lifting. ETL remains right when the transformation needs code the warehouse cannot run (complex parsing, machine learning features, external enrichment), when data must be reduced before it lands (cost, privacy), or when the destination is not a query engine.

In practice most platforms are both: a thin extract-and-load layer that lands raw data faithfully, a SQL transformation layer for modelling, and code-based jobs for the few things SQL cannot do. The principle that matters more than the acronym is load raw, transform explicitly: keep the source's shape at the first hop, so every later step is reproducible.

Extraction: incremental, with watermarks

Pulling an entire table every night works until the table is large. Incremental extraction pulls only what changed since the last run, tracked by a watermark: the maximum updated_at (or id, or log position) seen in the previous successful run. Each run selects rows past the watermark, loads them, and advances the watermark only after the load commits. Three rules keep it honest: overlap the window slightly to catch rows with equal timestamps or clock skew (and deduplicate on load), store the watermark with the run's state rather than in someone's head, and choose a watermark column the source actually maintains — an updated_at that some code paths forget to set will lose rows silently.

from datetime import datetime, timedelta, timezone

OVERLAP = timedelta(minutes=5)


def extract_incremental(conn, state, table: str, ts_col: str = "updated_at"):
    since = state.get_watermark(table) - OVERLAP          # re-read a little to be safe
    now = datetime.now(timezone.utc)
    rows = conn.execute(
        f"SELECT * FROM {table} WHERE {ts_col} > %s AND {ts_col} <= %s ORDER BY {ts_col}",
        (since, now),
    ).fetchall()
    return rows, now


def run(conn, state, loader):
    rows, new_watermark = extract_incremental(conn, state, "orders")
    loader.upsert("orders", rows, key="order_id")       # overlap rows collapse on the key
    state.set_watermark("orders", new_watermark)         # only after the upsert committed

Some sources cannot be extracted incrementally: no reliable timestamp, hard deletes that leave no trace, or small reference tables where a full pull is trivial. For those, full extracts with a snapshot comparison are simpler and safer. For high-volume transactional sources the better answer is change data capture from the database log, which a later module covers; the watermark pattern is the workhorse for APIs and for tables that maintain their timestamps.

Loading idempotently

The load is where re-runs go wrong. Three patterns make a load idempotent — safe to repeat with the same result. Partition overwrite: the run computes one logical partition (a day, an hour) and replaces it atomically; a retry replaces it again with the same content. Upsert (merge) keyed on a natural key: rows that exist are updated, new ones inserted, and the overlap from the extract collapses. Atomic swap: build the new table under a temporary name, validate it, then swap names in one transaction so readers never see a partial state. Choose by grain: partition overwrite for event-like data with a time partition, upsert for entity-like data that changes, swap for full rebuilds of small-to-medium tables.

-- 1. partition overwrite: the whole day is replaced in one transaction
BEGIN;
DELETE FROM fact_events WHERE event_date = DATE '2026-03-01';
INSERT INTO fact_events
SELECT * FROM stage_events WHERE event_date = DATE '2026-03-01';
COMMIT;
-- (warehouses: INSERT OVERWRITE PARTITION, or MERGE on a table format)

-- 2. build, validate, swap: readers see the old table until the instant of the swap
CREATE TABLE dim_product_new AS SELECT ... FROM stage_products;
-- validation queries run here: row count within 10% of current, no NULL keys, keys unique
BEGIN;
ALTER TABLE dim_product     RENAME TO dim_product_old;
ALTER TABLE dim_product_new RENAME TO dim_product;
COMMIT;
DROP TABLE dim_product_old;

Append-only loads are idempotent only if every row carries a unique id and the destination deduplicates, or if the pipeline records which batches were loaded and skips repeats. Plain INSERT of a batch with no key is the single most common source of duplicated data in warehouses; treat it as a bug in the design, not an operational risk to manage.

Late, changed and out-of-order data

Events arrive late: a mobile device syncs a day later, a partner sends yesterday's file this afternoon, a record is corrected a week after it was created. A pipeline that processes "yesterday" once and moves on will miss them. Two design choices handle this. Partition by event time, not arrival time, so a late event lands in the partition where it belongs; and reprocess a trailing window — recompute the last N days on every run, where N covers most lateness — so late arrivals are picked up automatically. Beyond the window, a manual backfill handles the rare very late record.

from datetime import date, timedelta

LOOKBACK_DAYS = 3


def partitions_to_rebuild(run_date: date) -> list[date]:
    return [run_date - timedelta(days=i) for i in range(LOOKBACK_DAYS)]


def daily(run_date: date, build_partition):
    for day in partitions_to_rebuild(run_date):
        build_partition(day)          # idempotent overwrite of that day's partition


# run on 2026-03-04: rebuilds 03-04, 03-03, 03-02 from the raw layer
daily(date(2026, 3, 4), build_partition=lambda d: print("rebuilding", d))

Updates to existing records are the same problem with a different face: an order's status changes from placed to paid. For entity tables, upsert on the key. For fact tables that must not lose history, load status changes as new events and derive the current state with a window function, or maintain an accumulating snapshot as the modelling module described. Whichever you choose, write down how each table treats late and changed data; it is the first question anyone debugging a discrepancy will ask.

Backfills, reprocessing and operations

A backfill runs the pipeline for a range of past dates: after a bug fix, after a new column is added, when a new consumer needs history. Backfills are safe only if every step is idempotent and parameterised by logical date, which is why those properties were non-negotiable. Run backfills as a distinct operation with its own concurrency limit (they compete with the daily run for warehouse compute and source rate limits), from the raw layer rather than from the source system when possible, and validate a sample of rebuilt partitions against the previous version before dropping it.

• Logging and metrics per run: rows extracted, loaded, rejected; duration; watermark before and after. A run that loads zero rows on a Tuesday should alert.
• Freshness checks: the newest event time in each table versus now; the SLA is a number.
• Reconciliation: periodic counts or sums compared with the source for a window; the only way to catch silent loss.
• Small, single-purpose steps: extract, stage, transform, publish — each restartable, each with a clear input and output.
• Schema change handling: new columns added automatically to raw; type or rename changes fail loudly with a message that names the column.