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."""
    if not rows:            # nothing new since the watermark; executemany() rejects an empty list
        return
    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

Answer from memory before opening each one.

1. Rental 150 was loaded last week. Today the application changes its gpu_model in Postgres. With the extractor exactly as written in this chapter, what does the next run do about it — and doesn't the upsert take care of updates?

   Show answer

   Nothing: the watermark is max(rental_id), and 150 is below it, so the row is never fetched again. The tempting answer is "the upsert handles it", but the upsert can only update rows that the extract brings back. A watermark only captures what its column captures — an id sees new rows; to see changed rows you need a column that moves on every change, such as updated_at. See Pick a good watermark column.

2. A different team stores its watermark in a separate state file, written after the load. One night the load succeeds, then the job dies before the file is updated, and the scheduler retries. What does the retry fetch, and what decides whether the destination ends up correct?

   Show answer

   It fetches the same rows again, because the watermark still points at the previous run. With a plain INSERT every one of them is duplicated; with an upsert on the key (or a replace-by-partition) the second write changes nothing. The watermark limits what you fetch, the idempotent write makes a re-run harmless — they guard different failures, which is why you want both. See Idempotent loads.

3. You discover that a cleaning rule has been mangling region for three months. Compare the cost of the fix in an ETL pipeline and in an ELT one.

   Show answer

   In ETL only the transformed result was loaded, so the raw rows are gone: you must re-extract three months from the source — slow, a load on the live database, and the source may have changed since. In ELT the raw rows were loaded into bronze first, so you fix the SQL and re-run the transform in the warehouse against untouched data. Keeping load and transform separate is what buys that. See ELT vs ETL.

4. In the replace-by-partition pattern, the DELETE and the INSERT sit inside BEGIN … COMMIT. If they didn't, what would a dashboard reading bronze_rentals see when the job dies between the two statements?

   Show answer

   A day with no rentals at all — the partition was deleted and never refilled, and it stays that way until someone re-runs the job. Inside one transaction the pair is atomic: readers see either the old partition or the new one, never the gap. See Pattern B.

5. A partner sends one file per day containing all of that day's rentals. They sometimes re-issue a day's file with corrections — and with some rows removed. Which idempotent-load pattern do you choose, and when would the other one be the better fit?

   Show answer

   Replace-by-partition: delete that load_date and insert the re-issued file, so the day ends up exactly equal to the source. An upsert would fix the corrected rows but leave behind the ones the partner removed. Upsert is the better fit when rows trickle in incrementally, may be updates to existing keys, and there is no clean slice to reload wholesale — like the watermark extractor in this chapter. See Idempotent loads.

6. A product manager asks for the revenue dashboard to be "real-time" and suggests moving ingestion to streaming. The dashboard is read each morning. What do you propose, and what would have to be true for streaming to be the right call?

   Show answer

   Keep batch — hourly, or micro-batch every few minutes if fresher numbers are genuinely useful: it is simpler, cheaper and easier to reason about, and nobody reading a morning dashboard benefits from seconds of latency. Streaming earns its complexity only when something acts on the data within seconds; reaching for it earlier is a classic over-engineering trap. See Batch, micro-batch & streaming.

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
rows = cur.fetchall()

if rows:   # a second run finds nothing new, and executemany() rejects an empty list
    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)  # 2. idempotent
print(con.execute("SELECT count(*) FROM bronze_rentals").fetchone())

Takeaway

• Pick the extraction method from the source's shape; the goal is always to find what is new, pull it and land it.
• Incremental extraction with a high watermark moves only new rows and spares a source that is also serving live traffic.
• The watermark column must only ever go up: an auto-increment id suits append-only data, and updated_at suits rows that change after insert.
• ELT loads raw data first and transforms inside the warehouse, so fixing a logic bug means re-running SQL on data you already hold.
• A load is idempotent when running it twice leaves the same state as running it once; use an upsert on a key, or delete and reload one partition.
• The watermark prevents refetching and the idempotent write prevents duplicates; they guard different failures, so keep both.

A watermark only sees what its column reveals, so deletes and silent updates slip past it, and the next chapter captures every change from the database log. → Change Data Capture

Further reading

• PostgreSQL: INSERT … ON CONFLICT — PostgreSQL docs, accessed 2026. The upsert syntax behind the idempotent load; DuckDB uses the same form.
• DuckDB: INSERT — DuckDB docs, accessed 2026. The ON CONFLICT clause and its requirement for a unique constraint.
• Designing Data-Intensive Applications — Kleppmann, O'Reilly. Chapters on batch processing and derived data explain why re-runnable loads matter.