Your First Pipeline

What you'll build

Real data platforms all do the same fundamental thing: take data out of the systems that produce it, reshape it into something useful, and put it where people can analyze it. That three-step move has a name — ETL (extract, transform, load), sometimes ordered ELT. You're going to build a tiny but genuine one.

Our story: a GPU-rental marketplace records every rental in Postgres (the operational source). We want a business question answered — how much revenue did each GPU model earn? — so we'll pull the rentals into DuckDB (our analytics warehouse), compute revenue, aggregate by model, and save the answer as a clean table.

Seed the source

Before we can extract anything, the source needs data. Let's create a rentals table in Postgres and drop in eight sample rentals. Connect to your running Postgres with psql (the client is bundled in the Postgres container):

docker compose exec postgres psql -U griddp -d source

At the source=# prompt, paste this SQL:

DROP TABLE IF EXISTS rentals;

CREATE TABLE rentals (
    rental_id      INTEGER PRIMARY KEY,
    customer_id    INTEGER NOT NULL,
    gpu_model      TEXT    NOT NULL,
    hours          NUMERIC NOT NULL,
    price_per_hour NUMERIC NOT NULL,
    created_at     TIMESTAMP DEFAULT now()
);

INSERT INTO rentals (rental_id, customer_id, gpu_model, hours, price_per_hour) VALUES
    (1, 101, 'A100', 10.0, 2.50),
    (2, 102, 'H100', 5.0,  4.00),
    (3, 101, 'A100', 3.5,  2.50),
    (4, 103, 'L4',   20.0, 0.80),
    (5, 102, 'H100', 8.0,  4.00),
    (6, 104, 'A100', 1.5,  2.50),
    (7, 103, 'L4',   12.0, 0.80),
    (8, 101, 'H100', 6.0,  4.00);

SELECT * FROM rentals ORDER BY rental_id;

Extract — read from the source

Extraction means: connect to the source and pull the rows we need. We'll use psycopg to query Postgres and load the result straight into a pandas dataframe. In your mini-griddp folder, create extract.py:

import psycopg
import pandas as pd

PG_CONN = "host=localhost port=5432 dbname=source user=griddp password=griddp_pw"

def extract_rentals() -> pd.DataFrame:
    """Connect to Postgres and read the rentals table into a dataframe."""
    with psycopg.connect(PG_CONN) as conn:
        df = pd.read_sql("SELECT * FROM rentals", conn)
    return df

if __name__ == "__main__":
    rentals = extract_rentals()
    print(f"Extracted {len(rentals)} rows")
    print(rentals.head())

uv run python extract.py

Transform — reshape the data

Now the interesting part: turn raw rows into an answer. We need two steps — compute revenue = hours × price_per_hour for each rental, then sum that revenue grouped by gpu_model.

We have a choice of where to do this. Two common options:

We'll use DuckDB SQL. DuckDB can query a pandas dataframe directly as if it were a table, the transform reads almost exactly like the business question, and this is the pattern real warehouses use — let the analytics engine do the heavy lifting. Create transform.py:

import duckdb
import pandas as pd

def transform(rentals: pd.DataFrame) -> pd.DataFrame:
    """Compute revenue per rental, then aggregate to revenue per GPU model."""
    result = duckdb.sql("""
        SELECT
            gpu_model,
            COUNT(*)                        AS rental_count,
            SUM(hours * price_per_hour)     AS total_revenue
        FROM rentals
        GROUP BY gpu_model
        ORDER BY total_revenue DESC
    """).df()
    return result

if __name__ == "__main__":
    from extract import extract_rentals
    summary = transform(extract_rentals())
    print(summary)

uv run python transform.py

Load — land it in the warehouse

The answer exists, but only in memory — when the script ends it's gone. Loading means persisting the result somewhere durable. We'll write it into a DuckDB database file as a table called gpu_model_revenue, and also drop a Parquet file (the standard columnar format you'll meet everywhere later). Create load.py:

import duckdb
import pandas as pd

WAREHOUSE = "warehouse.duckdb"

def load(summary: pd.DataFrame) -> None:
    """Persist the summary into the DuckDB warehouse and to Parquet."""
    con = duckdb.connect(WAREHOUSE)
    con.register("summary_df", summary)
    con.sql("CREATE OR REPLACE TABLE gpu_model_revenue AS SELECT * FROM summary_df")
    con.sql("COPY gpu_model_revenue TO 'gpu_model_revenue.parquet' (FORMAT parquet)")
    con.close()

if __name__ == "__main__":
    con = duckdb.connect(WAREHOUSE)
    print(con.sql("SELECT * FROM gpu_model_revenue").df())
    con.close()

That __main__ block lets you query the warehouse back after the pipeline has run — proving the data is really on disk:

uv run python load.py   # (run this after the full pipeline below)

Run end to end

Three functions, one flow. Let's assemble them into a single pipeline.py that runs the whole extract → transform → load in one command:

from extract import extract_rentals
from transform import transform
from load import load

def run() -> None:
    print("→ Extracting from Postgres…")
    rentals = extract_rentals()
    print(f"  got {len(rentals)} rows")

    print("→ Transforming…")
    summary = transform(rentals)

    print("→ Loading into DuckDB warehouse…")
    load(summary)

    print("✓ Pipeline complete. Result:")
    print(summary)

if __name__ == "__main__":
    run()

uv run python pipeline.py

A first big idea: make it re-runnable

Run uv run python pipeline.py a second time. Notice the result is exactly the same — not doubled. That's not luck; it's by design, and it's one of the most important properties a pipeline can have.

The magic word is idempotent: running the operation twice leaves the system in the same state as running it once. Press an elevator button twice — same result. Our load uses CREATE OR REPLACE TABLE, which rebuilds the table from scratch every run instead of appending to it. So a re-run can't produce duplicates.

Reflect — map it to the mental model

Remember the data-platform layers from Chapter 01? You just built one of each:

That's a real platform's spine. But it's a toy, and the gap between this and production is exactly what the rest of the curriculum fills. Here's what's still missing:

Commit your work

You built something real — get it into version control so it's tracked and part of your portfolio. From inside mini-griddp:

# don't commit the generated warehouse/parquet — they're rebuildable
echo "warehouse.duckdb" >> .gitignore
echo "*.parquet"        >> .gitignore

git add extract.py transform.py load.py pipeline.py .gitignore pyproject.toml uv.lock
git commit -m "Add first ETL pipeline: Postgres rentals -> DuckDB revenue"
git push

✓ Check yourself

Answer from memory before opening each one.

1. transform.py already prints the revenue-per-model answer. What would be missing if the pipeline stopped there, and what does the load step add?

   Show answer

   The answer would exist only in memory and vanish when the script ends — nobody else could use it without re-running everything against the source. Loading persists it as the gpu_model_revenue table in warehouse.duckdb (plus a Parquet file), so it can be queried at any time. That durable, queryable result is what makes it a warehouse rather than a printout. See Load.

2. Someone changes load.py to create the table once and then INSERT INTO gpu_model_revenue SELECT * FROM summary_df on every run. The pipeline runs three times in a day. How many rows does the table hold, and what does a report that sums total_revenue show?

   Show answer

   Nine rows — three copies of each model's row — and the report shows three times the true revenue. Nothing errors; the numbers are just wrong. The load stopped being idempotent: CREATE OR REPLACE TABLE rebuilds the table each run, so a re-run leaves the same state, whereas a blind insert stacks another copy every time. See Make it re-runnable.

3. A ninth rental lands in Postgres — (9, 105, 'L4', 10.0, 0.80) — and you re-run pipeline.py unchanged. How many rows does gpu_model_revenue have now, and what does the L4 row say?

   Show answer

   Still three rows, one per GPU model; L4 now shows rental_count 3 and total_revenue 33.6 (25.6 + 10 × 0.80), and it stays in third place. Each run extracts the whole rentals table and rebuilds the summary from scratch, so new source rows flow through without any old ones being counted twice. See Transform and Make it re-runnable.

4. The commit includes uv.lock, which is machine-generated, but leaves out warehouse.duckdb and the Parquet file, which are also machine-generated. What is the rule that separates them?

   Show answer

   Commit what is needed to rebuild; ignore what gets rebuilt. uv.lock is an input — without it a teammate can't recreate the same environment. The warehouse file and the Parquet are outputs: anyone with the code and the source can regenerate them by running the pipeline, so committing them only adds bulk and stale data. "Generated" isn't the test; "reproducible from what's in the repo" is. See Commit your work.

5. The chapter did the transform in DuckDB SQL, though pandas could have done it. A colleague prefers pandas for everything. What decides between them, and when is pandas the better call?

   Show answer

   Size and the kind of logic. Filter–join–aggregate work is what an analytics engine is built for: the SQL reads like the business question, the engine does the heavy lifting, and the same pattern keeps working as the data grows. Pandas wins when the data is small and the logic is genuinely Python-shaped — custom functions, awkward parsing, calling a library per row. At 8 rows either is fine; the reason to practise the SQL route is that it is the one that scales. See Transform.

6. Rebuilding gpu_model_revenue from scratch on every run is the simplest way to stay idempotent. What about the real GridDP would eventually make that a poor choice, and which alternative named in this chapter keeps re-runs safe?

   Show answer

   Volume. A full rebuild re-reads every rental ever made on every run — trivial at 8 rows, slow and wasteful at millions when only today's rows changed. An upsert (update the rows that exist, insert the ones that don't) touches only what changed and is still safe to repeat, at the price of needing a reliable key and more careful logic. While the table is small, keep the rebuild: simple and obviously correct beats clever. See the trap we avoided and what's still missing.

def transform_customer_hours(rentals):
    import duckdb
    return duckdb.sql("""
        SELECT customer_id, SUM(hours) AS total_hours
        FROM rentals
        GROUP BY customer_id
        ORDER BY total_hours DESC
    """).df()

def load_table(summary, table_name):
    con = duckdb.connect(WAREHOUSE)
    con.register("df", summary)
    # CREATE OR REPLACE = idempotent: rebuilds, never appends
    con.sql(f"CREATE OR REPLACE TABLE {table_name} AS SELECT * FROM df")
    con.close()

from extract import extract_rentals
from transform import transform, transform_customer_hours
from load import load_table

def run():
    rentals = extract_rentals()
    load_table(transform(rentals), "gpu_model_revenue")
    load_table(transform_customer_hours(rentals), "customer_hours")
    print("✓ Loaded gpu_model_revenue and customer_hours")

if __name__ == "__main__":
    run()

Takeaway

• A pipeline is extract, transform, load: read from the source system, reshape the data, and persist the result where people can query it.
• Pushing the aggregation into DuckDB SQL keeps the transform close to the business question and scales better than looping in Python.
• A pipeline is idempotent when running it twice leaves the same state as running it once. CREATE OR REPLACE TABLE gives that, while a bare INSERT INTO stacks duplicates on every rerun.
• Retries, overlapping schedules and manual debugging runs all happen, so idempotency is what makes "just run it again" safe.
• Commit the code that builds the warehouse and ignore the generated database and Parquet files.
• The toy lacks scheduling, tests, scale, many sources and safe secrets, and each later course adds one of them to this same extract, transform, load shape.

You built the what; Course 3 explains why it works and how it scales. → Foundations

Further reading

• psycopg 3 documentation — Psycopg project. Connecting to Postgres and running queries from Python.
• DuckDB Python client overview — DuckDB documentation. Connections, persistent database files and the .df() conversion.
• SQL on pandas — DuckDB documentation. How DuckDB reads a DataFrame by name from the Python scope.
• CREATE TABLE — DuckDB documentation. CREATE OR REPLACE TABLE, the statement that makes the load rerunnable.
• INSERT, including ON CONFLICT — PostgreSQL documentation. The upsert alternative to replacing the whole table.