Lab 1 — Model a pipeline end to end

Objective

By the end of this lab you will have a runnable Python script, pipeline.py, that ingests a day of raw order events, walks them through the five lifecycle stages you learned in this course, and serves a summary.csv of revenue by category — with a quality check that refuses to serve garbage. It uses only the Python standard library, so it costs nothing and runs anywhere.

Skills exercised:

• Mapping a real dataset onto the generation-to-serving lifecycle
• Implementing an ingestion stage that applies a batch (target-day) decision
• Writing a transformation that deduplicates and aggregates correctly
• Adding a data-quality undercurrent that fails loudly

Prerequisites and setup

Prerequisites: the three lessons in this module — What is data engineering, The data engineering lifecycle, and Batch vs. streaming.

Setup: you need Python 3.10 or newer. No packages to install; everything uses the standard library. Verify your version and create a working folder.

macOS / Linux

python3 --version
mkdir de101-pipeline-lab && cd de101-pipeline-lab

Windows (PowerShell)

python --version
mkdir de101-pipeline-lab; cd de101-pipeline-lab

Expected output from the version command is a line like Python 3.11.6. If the command is not found, install Python from python.org before continuing. In the steps below, use python3 on macOS/Linux and python on Windows.

note

Every step writes or edits files in this one folder. If you close your terminal and come back, re-run the cd de101-pipeline-lab command to return to it; all your files are still there.

Step 1 — Create the source data and load it

Create the raw dataset (the generation stage's output, handed to you) and a function to read it from storage.

Save this exactly as orders.csv in your lab folder:

order_id,created_at,category,amount,status
1001,2026-07-16T09:12:00,books,20.00,completed
1002,2026-07-16T10:05:00,books,15.50,completed
1003,2026-07-16T11:30:00,electronics,299.00,completed
1004,2026-07-16T12:00:00,electronics,299.00,refunded
1002,2026-07-16T10:05:00,books,15.50,completed
1005,2026-07-15T23:59:00,books,9.99,completed

That file has deliberate messiness you will handle later: order 1002 appears twice, 1004 was refunded, and 1005 is from the previous day.

Now create pipeline.py with the first function:

import csv
from collections import defaultdict
from datetime import date, datetime


def read_orders(path):
    """Storage read: load raw order rows from CSV into dicts."""
    with open(path, newline="") as f:
        return list(csv.DictReader(f))

Checkpoint: run the loader and count the rows.

python3 -c "import pipeline; print(len(pipeline.read_orders('orders.csv')))"

You should see 6. If you see 5, your orders.csv is missing the duplicate 1002 row; if you see an error, check the file is named exactly orders.csv in the current folder.

Step 2 — Ingest with a batch target-day decision

Goal: keep only the rows for the day you are processing. This is the ingestion stage, and choosing "one day at a time" is a batch decision.

Add this function to pipeline.py:

def ingest(orders, target_day):
    """Ingestion (batch): keep only rows created on the target day."""
    kept = []
    for row in orders:
        created = datetime.fromisoformat(row["created_at"]).date()
        if created == target_day:
            kept.append(row)
    return kept

Checkpoint: ingest for July 16 and count the survivors.

python3 -c "import pipeline; from datetime import date; \
raw = pipeline.read_orders('orders.csv'); \
print(len(pipeline.ingest(raw, date(2026, 7, 16))))"

You should see 5 — the five July 16 rows, with the July 15 row (1005) correctly filtered out. If you see 6, the date filter is not applying; if you see 0, check that target_day is date(2026, 7, 16).

Step 3 — Transform: drop refunds, dedupe, aggregate

Goal: turn the ingested rows into revenue per category. This transformation stage carries the correctness logic — it drops non-completed orders, deduplicates by order_id, and sums amount per category.

Add this function:

def transform(orders):
    """Transformation: keep completed, dedupe by order_id, sum by category."""
    seen = set()
    revenue = defaultdict(float)
    for row in orders:
        if row["status"] != "completed":
            continue
        if row["order_id"] in seen:
            continue
        seen.add(row["order_id"])
        revenue[row["category"]] += float(row["amount"])
    return dict(revenue)

Checkpoint: run the full ingest-then-transform on July 16.

python3 -c "import pipeline; from datetime import date; \
raw = pipeline.read_orders('orders.csv'); \
day = pipeline.ingest(raw, date(2026, 7, 16)); \
print(pipeline.transform(day))"

You should see exactly {'books': 35.5, 'electronics': 299.0}. The refunded 1004 is dropped and the duplicate 1002 is counted once. If electronics shows 598.0, the refund filter is missing; if books shows 51.0, the dedupe is not working.

Step 4 — Add the data-quality undercurrent

Goal: never serve nonsense. This quality undercurrent rejects an empty result or any negative total before it can reach a consumer.

Add this function:

def quality_check(revenue):
    """Quality undercurrent: refuse empty or negative revenue."""
    if not revenue:
        raise ValueError("No revenue rows — refusing to serve an empty summary")
    for category, amount in revenue.items():
        if amount < 0:
            raise ValueError(f"Negative revenue for {category}: {amount}")
    return revenue

Checkpoint: confirm it passes good data and blocks empty data.

python3 -c "import pipeline; \
print(pipeline.quality_check({'books': 35.5})); \
print('---'); \
pipeline.quality_check({})"

You should see {'books': 35.5}, then ---, then a ValueError with the message about refusing an empty summary. Seeing the error here means the guard works — that is the intended result, not a failure.

Step 5 — Serve the summary and assemble the pipeline

Goal: write the consumer-facing output (serving) and wire all five stages into one runnable pipeline.

Add these two functions to finish pipeline.py:

def serve(revenue, path):
    """Serving: write the summary consumers will read."""
    with open(path, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["category", "revenue"])
        for category in sorted(revenue):
            writer.writerow([category, f"{revenue[category]:.2f}"])


def run_pipeline(source, target_day, out):
    raw = read_orders(source)
    ingested = ingest(raw, target_day)
    revenue = quality_check(transform(ingested))
    serve(revenue, out)
    return revenue


if __name__ == "__main__":
    result = run_pipeline("orders.csv", date(2026, 7, 16), "summary.csv")
    print(result)

Checkpoint: run the whole file, then inspect the output it wrote.

python3 pipeline.py
cat summary.csv

The first command prints {'books': 35.5, 'electronics': 299.0}. The summary.csv file then contains a header plus two category rows. On Windows PowerShell, use Get-Content summary.csv instead of cat.

Validation

Verify the whole build against the objective: a served summary that is correct end to end. Run this validation one-liner, which reruns the pipeline and checks the served file's contents:

python3 -c "import pipeline; from datetime import date; \
pipeline.run_pipeline('orders.csv', date(2026, 7, 16), 'summary.csv'); \
rows = pipeline.read_orders('summary.csv'); \
ok = rows == [{'category': 'books', 'revenue': '35.50'}, \
{'category': 'electronics', 'revenue': '299.00'}]; \
print('PASSED — pipeline complete' if ok else 'FAILED — check the summary')"

Checkpoint: the command prints PASSED — pipeline complete. If it prints FAILED, open summary.csv: books must be 35.50 and electronics must be 299.00, sorted alphabetically by category.

Teardown

Nothing is billable — this lab runs entirely on your machine. Keep the de101-pipeline-lab folder if you want to reuse pipeline.py as a reference; otherwise delete it with rm -rf de101-pipeline-lab (macOS/Linux) or Remove-Item -Recurse de101-pipeline-lab (Windows).

Troubleshooting

Symptom	Likely cause	Fix
ModuleNotFoundError: No module named 'pipeline'	Running from the wrong folder	cd into de101-pipeline-lab, where pipeline.py lives
FileNotFoundError: 'orders.csv'	Data file missing or misnamed	Re-save the Step 1 CSV as exactly orders.csv in the lab folder
Loader prints 5 in Step 1	Duplicate 1002 row omitted	Add the second 1002 line from the Step 1 data
electronics totals 598.0 in Step 3	Refunded row not filtered	Keep only rows where status == "completed" in transform
books totals 51.0 in Step 3	Duplicate not deduplicated	Confirm the seen set is checked and updated by order_id
Validation prints FAILED	Wrong rounding or category order	serve must format with f"{value:.2f}" and sort categories