Data Engineering Fundamentals

A data pipeline moves data from where it is produced to where it is asked questions. This lesson builds a complete one — ingest, validate, transform, serve — and then breaks it in the three ways that matter.

The stages

Everything else in this track is one of those six done properly.

Setup

pip install duckdb pandas pyarrow

Successfully installed duckdb-1.1.3 pandas-2.2.3 pyarrow-18.1.0

Two source files, the kind a loader would drop into a landing directory:

# landing/customers.csv
customer_id,full_name,country_code,created_at
1,Ada Lovelace,GB,2025-11-02
2,Grace Hopper,US,2025-11-04
3,Alan Turing,GB,2026-01-03
4,Katherine Johnson,US,2026-01-08

# landing/orders_2026-01-04.csv
order_id,customer_id,ordered_at,status,amount
1001,1,2026-01-04,completed,25.50
1002,2,2026-01-04,completed,12.00
1003,1,2026-01-04,returned,40.00
1004,3,2026-01-04,completed,8.75
1005,2,2026-01-04,pending,63.20
1006,9,2026-01-04,completed,19.99

The whole pipeline

# pipeline.py
import duckdb
import sys

RUN_DATE = sys.argv[1] if len(sys.argv) > 1 else "2026-01-04"
con = duckdb.connect("bookshop.duckdb")

# ---- 1. ingest: land the file as-is, nothing dropped -------------------
con.execute("create schema if not exists bronze")
con.execute(f"""
    create or replace table bronze.orders as
    select *, '{RUN_DATE}' as _run_date, now() as _ingested_at
    from read_csv('landing/orders_{RUN_DATE}.csv', header = true,
                  columns = {{'order_id': 'BIGINT', 'customer_id': 'BIGINT',
                             'ordered_at': 'DATE', 'status': 'VARCHAR',
                             'amount': 'DECIMAL(10,2)'}})
""")
con.execute("create or replace table bronze.customers as select * from read_csv_auto('landing/customers.csv')")

# ---- 2. validate: refuse to continue on a broken load ------------------
checks = con.execute("""
    select
        count(*)                                        as rows,
        count(*) filter (order_id is null)              as null_ids,
        count(*) - count(distinct order_id)             as duplicate_ids,
        count(*) filter (amount < 0)                    as negative_amounts
    from bronze.orders
""").fetchone()

rows, null_ids, dup_ids, negatives = checks
print(f"ingested {rows} rows | nulls={null_ids} dups={dup_ids} negatives={negatives}")
if null_ids or dup_ids or negatives:
    sys.exit(f"validation failed: {checks}")

# ---- 3. transform: clean, then apply business rules --------------------
con.execute("create schema if not exists silver")
con.execute("""
    create or replace table silver.orders as
    select order_id, customer_id, ordered_at,
           lower(trim(status)) as status, amount
    from bronze.orders
    where status != 'pending'
""")

con.execute("create schema if not exists gold")
con.execute("""
    create or replace table gold.daily_revenue as
    select o.ordered_at, c.country_code,
           count(*) as orders, sum(o.amount) as revenue
    from silver.orders o
    join bronze.customers c using (customer_id)
    where o.status = 'completed'
    group by 1, 2
    order by 1, 2
""")

# ---- 4. serve ----------------------------------------------------------
print(con.execute("select * from gold.daily_revenue").df().to_string(index=False))

python pipeline.py 2026-01-04

ingested 6 rows | nulls=0 dups=0 negatives=0
ordered_at country_code  orders  revenue
2026-01-04           GB       2    34.25
2026-01-04           US       1    12.00

That is a real pipeline: six rows in, an aggregate out, with a gate in the middle that stops a broken load reaching the marts.

Break 1: the row that disappeared

Six orders arrived, but the revenue covers three. Where did the others go?

print(con.execute("""
    select 'bronze' as layer, count(*) from bronze.orders
    union all select 'silver', count(*) from silver.orders
    union all select 'in gold', sum(orders) from gold.daily_revenue
""").df().to_string(index=False))

  layer  count
 bronze      6
 silver      5
in gold      3

Six became five became three. Two of those drops are intended — pending is filtered in silver, returned is excluded from revenue. The third is not:

print(con.execute("""
    select o.order_id, o.customer_id, o.status
    from silver.orders o
    left join bronze.customers c using (customer_id)
    where c.customer_id is null
""").df().to_string(index=False))

 order_id  customer_id    status
     1006            9 completed

Order 1006 belongs to customer 9, who does not exist, so the inner join silently dropped £19.99 of revenue. An inner join is a filter. Every join in a pipeline should either be a left join or be accompanied by a check that it dropped nothing:

orphans = con.execute("""
    select count(*) from silver.orders o
    left join bronze.customers c using (customer_id)
    where c.customer_id is null
""").fetchone()[0]
if orphans:
    print(f"WARNING: {orphans} orders reference a missing customer")

WARNING: 1 orders reference a missing customer

Break 2: running it twice

python pipeline.py 2026-01-04
python pipeline.py 2026-01-04

ingested 6 rows | nulls=0 dups=0 negatives=0
ordered_at country_code  orders  revenue
2026-01-04           GB       2    34.25
2026-01-04           US       1    12.00

ingested 6 rows | nulls=0 dups=0 negatives=0
ordered_at country_code  orders  revenue
2026-01-04           GB       2    34.25
2026-01-04           US       1    12.00

Identical — this pipeline is idempotent, because every step uses create or replace. Now change ingestion to append, as a “faster” incremental version would:

con.execute(f"insert into bronze.orders select *, '{RUN_DATE}', now() from read_csv(...)")

ingested 6 rows  | ...
ingested 12 rows | nulls=0 dups=6 negatives=0
validation failed: (12, 0, 6, 0)

The validation caught it. Without that duplicate check the revenue would simply have doubled, with every command reporting success. Pipelines are re-run constantly — after a failure, after a fix, during a backfill — so “what happens if this runs twice” is the first question to ask of any pipeline step. Lesson 3 is entirely about it.

Break 3: the silent schema change

The source team adds a channel column and renames amount to amount_gbp:

order_id,customer_id,ordered_at,status,amount_gbp,channel
1007,1,2026-01-05,completed,31.20,web

duckdb.duckdb.BinderException: Binder Error: Column "amount" not found in FROM clause!
Candidate bindings: "amount_gbp"

A loud failure, because the ingest declared explicit column types. Had it used read_csv_auto, amount would have arrived as NULL for every new row and revenue would have quietly fallen to zero.

Prefer the pipeline that crashes. A pipeline that fails is fixed in an hour; one that produces plausible wrong numbers is discovered a quarter later, if at all.

ELT, and why raw data is kept

Notice the shape: land raw (bronze), clean (silver), aggregate (gold), and never transform on the way in. That is ELT, and the reason is recovery — a bug in the silver logic is fixed by rerunning silver, because bronze still holds exactly what arrived. Transform before landing and the original is gone.

Practice

orphans = con.execute("""
    select count(*) from bronze.orders o
    left join bronze.customers c using (customer_id)
    where c.customer_id is null
""").fetchone()[0]
if orphans:
    sys.exit(f"validation failed: {orphans} orphaned orders")

ingested 6 rows | nulls=0 dups=0 negatives=0
validation failed: 1 orphaned orders

  layer  count
 bronze      6
 silver      5
in gold      3

ingested 12 rows | nulls=0 dups=6 negatives=0
validation failed: (12, 0, 6, 0)

duckdb.duckdb.BinderException: Binder Error: Column "amount" not found in FROM clause!
Candidate bindings: "amount_gbp"

Next: file formats — why the same data is 4× smaller and 30× faster to query as Parquet.