Lesson 5 - Guided Project: CityFlow's ETL Job

Welcome to the Guided Project#

This is CityFlow’s first real Dockerfile — not a toy script, a genuine, recognizable slice of the zone-hour ETL pipeline Course 1 built in pandas: stream the file a row group at a time, keep three columns of nineteen, aggregate into a per-zone, per-hour report. You’ll run it against the real January 2024 trip data, apply every technique from this module, and measure the real size difference between an unoptimized build and a properly slimmed one.

By the end of this project, you will be able to:

• Containerize a real, multi-stage-of-thinking data job, choosing correctly which techniques from this module actually apply.
• Mount a real dataset into a container at run time instead of baking it into the image, and explain why that matters.
• Apply the batch-job exit-code convention from Lesson 4 to a job with genuine failure modes.
• Measure a real, complete before-and-after image size, and account for where the savings came from.

The job: CityFlow’s zone-hour ETL, containerized#

This is the same shape as the out-of-core pipeline from Scaling Python for Data Engineering: stream the file a row group at a time — never load it whole — keep only the columns the report needs, and aggregate into a per-zone, per-hour trip-count and revenue report. This run covers one real month rather than the full quarter; the point here is containerizing the job correctly, not re-deriving the multi-month proof Course 1 already built.

"""CityFlow's zone-hour ETL job - this module's guided project, and this
codebase's very first Dockerfile for a real piece of CityFlow's pipeline.

Same shape as the out-of-core pipeline from Scaling Python for Data
Engineering: stream the file a row group at a time (never load it whole),
keep only the 3 of 19 columns the report needs, and aggregate into a
per-zone, per-hour trip-count and revenue report. This run covers one real
month (January 2024) rather than the full quarter - the point here is
containerizing the job correctly, not re-deriving the multi-month proof.

The dataset is NEVER baked into the image. DATA_PATH must point at a
mounted file (the normal case - fast, offline, reproducible); if nothing
is mounted, the job falls back to fetching the real public file itself.
Baking a ~50 MB dataset into the image would make every rebuild carry a
stale copy of data that changes independently of the code - exactly the
kind of coupling Lesson 1's "don't bake the data in" note warned against.

Exit codes follow the batch-job convention from Lesson 4:
  0  success            - report computed and printed
  2  no input available - nothing mounted AND the fallback fetch failed
  3  validation failed  - the file parsed but produced zero in-window trips
"""
import datetime as dt
import os
import sys
import urllib.request

import numpy as np
import pandas as pd
import pyarrow.parquet as pq

DATA_PATH = os.environ.get("DATA_PATH", "/data/trips.parquet")
FALLBACK_URL = (
    "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet"
)
KEEP = ["tpep_pickup_datetime", "PULocationID", "total_amount"]
WIN_LO = dt.datetime(2024, 1, 1)
WIN_HI = dt.datetime(2024, 2, 1)


def ensure_data():
    if os.path.exists(DATA_PATH):
        print(f"using mounted data at {DATA_PATH}", flush=True)
        return DATA_PATH
    print(f"no file at {DATA_PATH} - nothing mounted, fetching the real source", flush=True)
    tmp = "/tmp/trips.parquet"
    try:
        urllib.request.urlretrieve(FALLBACK_URL, tmp)
    except Exception as e:
        print(f"ERROR: no mounted data and fallback fetch failed: {e}", file=sys.stderr)
        sys.exit(2)
    return tmp


def aggregate_row_group(path, rg):
    tbl = pq.ParquetFile(path).read_row_group(rg, columns=KEEP)
    ts = tbl.column(0).to_numpy(zero_copy_only=False)
    keep = (ts >= np.datetime64(WIN_LO)) & (ts < np.datetime64(WIN_HI))
    hour = ts[keep].astype("datetime64[h]").astype("int64")
    zone = tbl.column(1).to_numpy()[keep]
    amt = tbl.column(2).to_numpy(zero_copy_only=False)[keep]
    part = (
        pd.DataFrame({"zone": zone, "hour": hour, "amt": amt})
        .groupby(["zone", "hour"], sort=False)
        .agg(trips=("amt", "size"), revenue=("amt", "sum"))
        .reset_index()
    )
    return part, int((~keep).sum())


def main():
    path = ensure_data()
    pf = pq.ParquetFile(path)
    parts, quarantined = [], 0
    for rg in range(pf.metadata.num_row_groups):
        part, q = aggregate_row_group(path, rg)
        parts.append(part)
        quarantined += q

    agg = pd.concat(parts, ignore_index=True)
    report = (
        agg.groupby(["zone", "hour"], sort=False)
        .agg(trips=("trips", "sum"), revenue=("revenue", "sum"))
        .reset_index()
    )

    if len(report) == 0 or report["trips"].sum() == 0:
        print("ERROR: zero in-window trips after aggregation", file=sys.stderr)
        sys.exit(3)

    print("CityFlow zone-hour ETL - January 2024")
    print("---------------------------------------")
    print(f"source rows (on disk):     {pf.metadata.num_rows:,}")
    print(f"zone-hour buckets:         {len(report):,}")
    print(f"trips kept:                {int(report['trips'].sum()):,}")
    print(f"quarantined (out-of-window): {quarantined}")
    print(f"total revenue:             ${report['revenue'].sum():,.2f}")

    top = report.loc[report["trips"].idxmax()]
    hour_str = str(np.array(int(top["hour"]), dtype="datetime64[h]"))
    print(
        f"busiest zone-hour:         zone {int(top['zone'])} at {hour_str}, "
        f"{int(top['trips'])} trips"
    )
    print("ETL completed successfully")
    sys.exit(0)


if __name__ == "__main__":
    main()

Save this as cityflow_etl.py, with requirements.txt naming both real dependencies:

pandas==2.2.3
pyarrow==19.0.0

The design decision: mount the data, never bake it in#

Notice cityflow_etl.py never COPYs a dataset into the image. It reads from DATA_PATH, defaulting to /data/trips.parquet — a path that only exists if something mounts a real file there at run time. If nothing is mounted, it falls back to fetching the real public file itself, from the same official NYC TLC source every course in this path uses.

This is a deliberate choice, not an oversight. Baking a real dataset into an image looks convenient — COPY trips.parquet . and you’re done — but it couples two things that should stay independent: the code, which changes when the ETL logic changes, and the data, which changes every month a new file arrives. Bake the data in, and every new month means rebuilding, retagging, and re-pushing the entire image just to reflect new rows the logic never touched. Mount it instead, and the same image runs correctly against January, February, or a year from now, unchanged.

docker run --rm -v /Users/you/datatweets/yellow_tripdata_2024-01.parquet:/data/trips.parquet:ro cityflow-etl-job:after

using mounted data at /data/trips.parquet
CityFlow zone-hour ETL - January 2024
---------------------------------------
source rows (on disk):     2,964,624
zone-hour buckets:         77,530
trips kept:                2,964,606
quarantined (out-of-window): 18
total revenue:             $79,455,941.50
busiest zone-hour:         zone 161 at 2024-01-24T18, 726 trips
ETL completed successfully

Real numbers, off the real file: 2,964,624 rows on disk, 77,530 distinct zone-hour buckets, 2,964,606 trips kept, 18 quarantined for falling outside the January window — the same stray out-of-window timestamps Course 1’s capstone found in this exact file — and $79,455,941.50 in total revenue. Confirm the dataset genuinely isn’t inside the image:

docker run --rm cityflow-etl-job:after find / -maxdepth 3 -iname "*trips*" -o -iname "*tripdata*"

No output — the file is nowhere in the image’s filesystem. It only exists inside the container because -v put it there, for this one run.

Applying Lesson 4: exit codes for a real ETL job#

The three exit codes from Lesson 4 apply directly here, and all three are real, reachable states for this exact job — not invented for the lesson:

docker run --rm --network none cityflow-etl-job:after

no file at /data/trips.parquet - nothing mounted, fetching the real source
ERROR: no mounted data and fallback fetch failed: <urlopen error [Errno -3] Temporary failure in name resolution>

ExitCode=2 — nothing mounted, and --network none genuinely blocks the fallback fetch too, exactly the “no data available at all” case exit code 2 is for.

docker run --rm -v "$(pwd)/out_of_window_trips.parquet":/data/trips.parquet:ro cityflow-etl-job:after

using mounted data at /data/trips.parquet
ERROR: zero in-window trips after aggregation

ExitCode=3 — a real, mounted parquet file with real rows, all dated outside January, so the aggregation runs cleanly and finds nothing to report. That’s a genuinely different failure from “no data at all,” and this job’s exit codes tell the two apart.

No HEALTHCHECK appears anywhere in this Dockerfile. That’s Lesson 4’s distinction, applied rather than just remembered: this is a batch job that runs once and stops, and a HEALTHCHECK — designed for a process that keeps running — would be a category error here. The exit code is the correct, complete signal.

Slimming it: before and after, measured#

Two Dockerfiles, same job, same script. Before applies nothing from this module — the full base image, COPY . . from a project folder that (realistically) has dev files sitting alongside the job:

FROM python:3.13

WORKDIR /app

COPY . .
RUN pip install --no-cache-dir -r requirements.txt

CMD ["python", "cityflow_etl.py"]

After applies Lessons 1 and 3 — the slim base, a real .dockerignore, dependencies installed before code is copied in:

FROM python:3.13-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY cityflow_etl.py .

CMD ["python", "cityflow_etl.py"]

docker build -f Dockerfile.before -t cityflow-etl-job:before .
docker build -f Dockerfile.after -t cityflow-etl-job:after .
docker images cityflow-etl-job

IMAGE                     ID             DISK USAGE   CONTENT SIZE   EXTRA
cityflow-etl-job:after    55f8a9e5183f        438MB             0B
cityflow-etl-job:before   5eea605c97a8       1.43GB             0B

1.43 GB down to 438 MB — 992 MB saved, a 3.3x reduction, and both images produce the exact same January report shown above, mounted data, byte for byte.

Two judgment calls this project makes explicitly#

No multi-stage build. Lesson 2 introduced multi-stage builds for a dependency that needed compiling. This job’s dependencies don’t: watch the pip install step in either Dockerfile’s build log, and every package downloads as a prebuilt .whl — pandas-2.2.3-cp313-cp313-manylinux2014_aarch64...whl, pyarrow-19.0.0-...whl — with no gcc invocation anywhere. Multi-stage only pays off when there’s a build toolchain to leave behind; applying it here would add Dockerfile complexity for zero measurable benefit, which is exactly the kind of unmeasured “best practice” this whole module has been arguing against.

No HEALTHCHECK. Already covered above, but worth stating as a decision rather than an omission: this is a run-to-completion batch job, and Lesson 4 drew a hard line between what a batch job needs (a meaningful exit code) and what a long-running service needs (a HEALTHCHECK). Adding one here wouldn’t just be unnecessary — it would be checking the health of a process that’s designed to finish and stop, which is a contradiction in terms.

Practice Exercises#

Exercise 1: Reproduce the full before-and-after measurement#

Build both Dockerfile.before and Dockerfile.after yourself, run each against the real, mounted January 2024 file, and confirm your own docker images output shows a real, multi-hundred-megabyte-to-gigabyte gap. Confirm both produce the identical report: 77,530 zone-hour buckets, 2,964,606 trips kept, $79,455,941.50 total revenue.

Exercise 2: Run it with no mount at all#

Run cityflow-etl-job:after with no -v flag and normal networking (not --network none). Confirm it falls back to fetching the real file itself and produces the identical report. Time how long the fetch adds compared to the mounted run.

Exercise 3: Extend the exit-code coverage#

This job currently has no explicit handling for a mounted file that isn’t valid Parquet at all (a corrupted or truncated download, for instance). Add a try/except around the parquet-reading step that catches that case and exits with a new code of your own choosing — document it in the module docstring the way 0/2/3 are documented, and justify why it deserves a code distinct from the existing three.

Summary#

CityFlow shipped its first real Dockerfile: a genuine, recognizable slice of Course 1’s zone-hour ETL logic — chunked ingest, dtype reduction, per-zone-per-hour aggregation — run against the real January 2024 trip data and producing real, verified numbers: 77,530 zone-hour buckets, 2,964,606 trips kept, 18 quarantined, and $79,455,941.50 in total revenue. The dataset was mounted at run time rather than baked into the image, with a genuine fallback fetch when nothing was mounted, and the job followed Lesson 4’s batch-job exit-code convention with all three codes reachable and demonstrated for real. Applying the slim base image and a real .dockerignore from Lessons 1 and 3 cut the image from 1.43 GB to 438 MB — 992 MB saved, a 3.3x reduction — while two other module techniques, a multi-stage build and a HEALTHCHECK, were deliberately left out, because neither one actually applied to this specific job, and applying them anyway would have been exactly the kind of unmeasured, reflexive choice this whole module argued against.

Key Concepts#

• Mount real data, don’t bake it in — keeps the image’s rebuild cadence independent of the dataset’s own update cadence.
• A fallback fetch makes an image correct in more environments — mounted for speed and reproducibility, self-fetching when nothing is provided.
• Exit codes for a real job’s real failure modes — success, no data available, and data present but invalid are three distinct, actionable states.
• Not every technique applies to every job — multi-stage builds and HEALTHCHECK both have real conditions for when they help; applying either without those conditions is cost with no benefit.
• Measure the whole thing, start to finish — the 992 MB, 3.3x number is only meaningful because both images were built, run, and verified against the same real data and the same real report.

Why This Matters#

This is the shape every real containerized data job takes: a slim, purpose-built image; data that arrives at run time rather than getting baked in; and failure modes that report themselves clearly enough for something else — a scheduler, an orchestrator, a human on call — to act on automatically. You’ve now watched CityFlow’s actual pipeline logic cross from “a pandas script on a laptop” to “a real, portable, measured container,” and every module for the rest of this course builds directly on that image: Module 4 puts it in a Compose stack alongside Postgres, and Module 5 onward runs it — and jobs shaped just like it — on a real local Kubernetes cluster.

Continue Building Your Skills#

That completes Module 3: Building Images for Data Jobs. You can choose a base image with a measured reason, split a build across stages when a dependency actually needs it, slim a bloated build context, give a batch job and a long-running service the check each one actually needs, and now you’ve done all of it at once on CityFlow’s real ETL logic.

Next comes Module 4: Docker Compose, where CityFlow’s job stops running alone. You’ll bring cityflow-etl-job and a real Postgres database together into a single, one-command multi-container stack — the first time this course’s containers talk to each other by name instead of by a manually created network, and the first time “run the whole local data stack” is a single command instead of a sequence of them.