Ingestion & ELT/ETL

The three source shapes

In Foundations you met the idea that data arrives in three fundamentally different shapes, and that each one needs a different way of getting at it. Before we move anything, let's recap — because how you extract is decided entirely by what the source is.

The mental model is the same in all three: find what's new, pull it, land it. Only the "how do I ask for new things" mechanic changes. This chapter works the database case end to end (it's the most common and the clearest), but the patterns transfer directly — the next chapter, Change Data Capture, sharpens the database case even further.

Full vs incremental extraction

There are exactly two ways to get rows out of a source, and choosing between them is the first real decision of any ingestion job.

• Full extract — pull every row, every run. Simple and always correct, but it re-reads data you already have. Fine for a 500-row reference table; ruinous for a 50-million-row events table you run hourly.
• Incremental extract — pull only the rows that are new or changed since last time. The workhorse of real pipelines.

The standard way to do an incremental extract is the high-watermark pattern. You keep a single value — the maximum id or updated_at you saw on the last run — and each new run asks the source for only rows above that mark. Then you save the new maximum for next time.

Why this matters so much:

• Cost & speed — you move kilobytes instead of gigabytes. A run that took an hour now takes seconds.
• You don't hammer the source — that Postgres is also serving the live application. A full table scan every 15 minutes competes with real customer traffic. A tight indexed range query barely registers.

ELT vs ETL

Both acronyms have the same three letters — Extract, Load, Transform — and differ only in the order. That order is the whole story.

ETL was born when storage was expensive and warehouses were weak. You couldn't afford to keep raw data, so you transformed it on a dedicated machine first and only loaded the polished result. The cost: every change to your logic meant re-extracting from the source, and your raw data was gone forever.

ELT flips the last two steps, and it won the industry for three concrete reasons:

This maps cleanly onto the medallion architecture you just built. Ingestion is the Load step: get raw rows into bronze, faithfully, with no cleaning. The Transform then happens later, in the warehouse, promoting bronze → silver → gold. Keeping load and transform separate is exactly what lets you re-run your cleaning logic against untouched raw data whenever you fix a bug. So: this chapter is the "EL" of ELT; chapter 07 is the "T".

Idempotent loads

Pipelines fail and get re-run — a network blip, a retry from your scheduler, a manual re-trigger after you fixed something. So a load step must be idempotent: running it twice with the same input leaves the destination in the same state as running it once. No duplicates. You met idempotency in Foundations and again in Course 2; here it becomes a daily habit.

The naive INSERT is not idempotent — re-run it and you get two copies of every row. There are two standard ways to fix that:

Pattern A — UPSERT / MERGE on a key. Insert new rows; for rows whose key already exists, update them instead of inserting again. Use this when rows arrive incrementally and may be updates.

-- DuckDB: requires a PRIMARY KEY or UNIQUE constraint on the key column
INSERT INTO bronze_rentals
SELECT * FROM staging_rentals
ON CONFLICT (rental_id) DO UPDATE SET
    customer_id = excluded.customer_id,
    gpu_model   = excluded.gpu_model,
    updated_at  = excluded.updated_at;
-- re-running with the same rows updates them in place -> no duplicates

Pattern B — replace by partition. Delete the slice you're about to load, then insert it fresh. Use this when you reload a whole day/hour at a time and want that window to be exactly the source.

BEGIN;
DELETE FROM bronze_rentals WHERE load_date = DATE '2026-06-21';
INSERT INTO bronze_rentals
SELECT * FROM staging_rentals WHERE load_date = DATE '2026-06-21';
COMMIT;
-- re-running replaces the same partition -> the result is identical

Hands-on: an incremental extractor

Let's build a real one: pull only new rentals from Postgres into a bronze table in DuckDB, using a high-watermark, idempotently. Then we'll run it twice and prove there are no duplicates.

Save this as extract_rentals.py in your mini-griddp repo:

import duckdb
import psycopg2

WAREHOUSE = "griddp.duckdb"           # local DuckDB file = our warehouse
PG = dict(host="localhost", port=5432, dbname="postgres",
          user="postgres", password="postgres")

def get_watermark(con):
    """The highest rental_id we've already loaded (0 if the table is empty)."""
    con.execute("""
        CREATE TABLE IF NOT EXISTS bronze_rentals (
            rental_id   BIGINT PRIMARY KEY,
            customer_id BIGINT,
            gpu_model   VARCHAR,
            updated_at  TIMESTAMP
        )""")
    row = con.execute("SELECT max(rental_id) FROM bronze_rentals").fetchone()
    return row[0] or 0

def extract_new_rows(watermark):
    """Pull ONLY rentals newer than the watermark from Postgres."""
    pg = psycopg2.connect(**PG)
    cur = pg.cursor()
    cur.execute(
        "SELECT rental_id, customer_id, gpu_model, updated_at "
        "FROM rentals WHERE rental_id > %s ORDER BY rental_id",
        (watermark,),
    )
    rows = cur.fetchall()
    pg.close()
    return rows

def load_idempotent(con, rows):
    """UPSERT on rental_id so a re-run can never duplicate."""
    con.executemany("""
        INSERT INTO bronze_rentals VALUES (?, ?, ?, ?)
        ON CONFLICT (rental_id) DO UPDATE SET
            customer_id = excluded.customer_id,
            gpu_model   = excluded.gpu_model,
            updated_at  = excluded.updated_at
    """, rows)

def main():
    con = duckdb.connect(WAREHOUSE)
    watermark = get_watermark(con)
    print(f"watermark (last rental_id loaded) = {watermark}")

    rows = extract_new_rows(watermark)
    print(f"pulled {len(rows)} new row(s) from Postgres")

    load_idempotent(con, rows)
    total = con.execute("SELECT count(*) FROM bronze_rentals").fetchone()[0]
    print(f"bronze_rentals now has {total} row(s)")
    con.close()

if __name__ == "__main__":
    main()

Now run it twice in a row, without changing the source in between:

python extract_rentals.py     # first run: loads everything
python extract_rentals.py     # second run: should load NOTHING new

To see the increment actually work, insert a couple of fresh rentals into Postgres and run the script a third time — it will report exactly those new rows and nothing else.

docker compose exec -T postgres psql -U postgres -c \
  "INSERT INTO rentals (customer_id, gpu_model, updated_at)
   VALUES (7, 'H100', now()), (7, 'A100', now());"
python extract_rentals.py     # -> pulled 2 new row(s); total grows by exactly 2

Batch, micro-batch & streaming

Our extractor runs, finishes, and exits — that's batch ingestion: move a chunk of data on a schedule (every hour, every night). It's by far the most common pattern, and for good reason — it's simple, cheap, and easy to reason about. Micro-batch is the same idea on a tight loop (every few seconds to minutes), trading a bit more cost for fresher data. Streaming processes each event the moment it arrives, continuously, with no "run" boundary at all — the right tool when seconds of latency genuinely matter. We build a real streaming pipeline in chapter 09; for now, just know that most of the ingestion you'll write is batch, and reaching for streaming before you need it is a classic over-engineering trap.

✓ Check yourself

• Can you name the three source shapes and how extraction differs for each?
• Can you explain the high-watermark pattern — what you store and how it filters the next pull?
• Why did ELT beat ETL? Can you tie its "L" and "T" to bronze and silver/gold?
• What makes a load idempotent, and what are the two ways to achieve it?

import duckdb, psycopg2

con = duckdb.connect("griddp.duckdb")
con.execute("""CREATE TABLE IF NOT EXISTS bronze_rentals
    (rental_id BIGINT, customer_id BIGINT, gpu_model VARCHAR, updated_at TIMESTAMP)""")

pg = psycopg2.connect(host="localhost", dbname="postgres",
                      user="postgres", password="postgres")
cur = pg.cursor()
cur.execute("SELECT rental_id, customer_id, gpu_model, updated_at FROM rentals")  # FULL pull
con.executemany("INSERT INTO bronze_rentals VALUES (?, ?, ?, ?)", cur.fetchall())  # plain INSERT
print(con.execute("SELECT count(*) FROM bronze_rentals").fetchone())

import duckdb, psycopg2

con = duckdb.connect("griddp.duckdb")
con.execute("""CREATE TABLE IF NOT EXISTS bronze_rentals
    (rental_id BIGINT PRIMARY KEY, customer_id BIGINT,
     gpu_model VARCHAR, updated_at TIMESTAMP)""")            # 1a. key for upsert

# 1b. high-watermark: highest id we already have (0 if empty)
watermark = con.execute("SELECT max(rental_id) FROM bronze_rentals").fetchone()[0] or 0

pg = psycopg2.connect(host="localhost", dbname="postgres",
                      user="postgres", password="postgres")
cur = pg.cursor()
cur.execute("SELECT rental_id, customer_id, gpu_model, updated_at "
            "FROM rentals WHERE rental_id > %s ORDER BY rental_id", (watermark,))  # 1c. incremental

con.executemany("""INSERT INTO bronze_rentals VALUES (?, ?, ?, ?)
    ON CONFLICT (rental_id) DO UPDATE SET
        customer_id = excluded.customer_id,
        gpu_model   = excluded.gpu_model,
        updated_at  = excluded.updated_at""", cur.fetchall())  # 2. idempotent
print(con.execute("SELECT count(*) FROM bronze_rentals").fetchone())

Next

The watermark only catches what its column reveals — it can miss deletes and silent updates. To capture every change a source makes, we go to the database's own change log. → Change Data Capture