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 griddp

At the griddp=# 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=griddp user=griddp password=griddp"

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

• Can you name the three stages of the pipeline and what each one does?
• Can you explain, in one sentence, what idempotent means and why it matters?
• Do you know why we committed the code but git-ignored warehouse.duckdb?
• Could you map your pipeline onto the source → ingestion → storage → transformation → serving layers?

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()

Next

That's a wrap on Course 2. Stop and appreciate what you can now do: you have a real data-engineering workstation, and you've built and committed a working pipeline that extracts from a source database, transforms data with SQL, and loads it into a warehouse — idempotently. That is the actual job, in miniature.

You built the what. Course 3 — Foundations explains the why: how databases really store and find data, how files and storage formats work under the hood, and how distributed systems let a pipeline like yours scale from 8 rows to billions. Once you understand the machinery beneath the tools, you stop following recipes and start engineering.

Course 3 — Foundations — is ready, and it's where you go next: it explains the machinery beneath everything you just built. Congratulations on finishing Course 2.