The same cleaning, every Monday
The owner sends a fresh export every Monday. It has the same faults as last week: repeats, words where numbers should be, odd spellings.
You could clean it by hand in a notebook each time. Or you can write the steps once, in order, in a script that runs the same way every time.
By the end of this lesson
Note
I can write a small pipeline that cleans, joins and reports, and stops when a check fails.
Key words
| Word | What it means | Example |
|---|---|---|
| Pipeline | steps run in order, each using the result of the one before | |
| Stage | one step of a pipeline | drop repeats |
| Check | a test inside the pipeline that can stop it | |
| Exit code | the number a script ends with, 0 for success | sys.exit(1) |
| Reproducible | giving the same result every time it is run |
The cells students fill in are left blank in this preview.
A pipeline is stages in a fixed order
A pipeline is a list of stages. Each one takes a table and gives back a table. The order matters: you cannot join on a date before the dates are clean.
After each stage the pipeline can print how many rows are left. A sudden drop is the first sign that a step went wrong.
Worked example: the stages of today's pipeline
| Stage | What it does | Rows expected |
|---|---|---|
| read | Loads the messy export | 367 |
| no repeats | Drops rows that appear twice | 365 |
| real counts | Keeps hires of 0 or more, as numbers | 363 |
| joined | Adds the holiday column, keeping every day | 363 |
The cells students fill in are left blank in this preview.
Write the holiday list again
docker exec jupyter sh -c "printf '%s\n' 'date,holiday' '2025-01-01,New Year' '2025-04-18,Good Friday' '2025-04-21,Easter Monday' '2025-05-05,Early May' '2025-05-26,Spring' '2025-08-25,Summer' '2025-12-25,Christmas' '2025-12-26,Boxing Day' > holidays.csv"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
The whole script is about twenty lines
The next block shows the script. Read it first. The function stage prints the rows left and stops the whole script with exit code 1 if there are fewer than least.
The number least is given when you run it, so you choose how strict the check is.
steps/pipeline.py
import sys
import pandas as pd
from mess import make
least = int(sys.argv[1])
def stage(name, frame):
print(name, "->", len(frame), "rows")
if len(frame) < least:
print("STOP:", name, "left fewer than", least, "rows")
sys.exit(1)
return frame
make()
frame = stage("read", pd.read_csv("messy.csv"))
frame = stage("no repeats", frame.drop_duplicates())
frame["hires"] = pd.to_numeric(frame["hires"], errors="coerce")
frame = stage("real counts", frame[frame["hires"] >= 0])
frame["weather"] = frame["weather"].str.strip().str.lower()
frame = stage("joined", frame.merge(pd.read_csv("holidays.csv"), on="date", how="left"))
frame.to_csv("report.csv", index=False)
print(frame.groupby("weather")["hires"].mean().round(1).to_string())
Write the pipeline into the lab
docker exec jupyter sh -c "printf '%s\n' 'import sys' 'import pandas as pd' 'from mess import make' '' 'least = int(sys.argv[1])' '' 'def stage(name, frame):' ' print(name, \"->\", len(frame), \"rows\")' ' if len(frame) < least:' ' print(\"STOP:\", name, \"left fewer than\", least, \"rows\")' ' sys.exit(1)' ' return frame' '' 'make()' 'frame = stage(\"read\", pd.read_csv(\"messy.csv\"))' 'frame = stage(\"no repeats\", frame.drop_duplicates())' 'frame[\"hires\"] = pd.to_numeric(frame[\"hires\"], errors=\"coerce\")' 'frame = stage(\"real counts\", frame[frame[\"hires\"] >= 0])' 'frame[\"weather\"] = frame[\"weather\"].str.strip().str.lower()' 'frame = stage(\"joined\", frame.merge(pd.read_csv(\"holidays.csv\"), on=\"date\", how=\"left\"))' 'frame.to_csv(\"report.csv\", index=False)' 'print(frame.groupby(\"weather\")[\"hires\"].mean().round(1).to_string())' > steps/pipeline.py"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Run it with a limit of 300 rows
docker exec jupyter python3 steps/pipeline.py 300Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
A check that stops the pipeline
A check is only useful if it can say no. Here the check is a limit on the rows left. Give it a limit the data cannot meet and the pipeline stops at the first stage that breaks it.
The script then ends with exit code 1, so whatever runs it knows it failed.
Ask for at least 365 rows
docker exec jupyter python3 steps/pipeline.py 365 || echo "exit code $?"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Ask for at least 400 rows
docker exec jupyter python3 steps/pipeline.py 400 || echo "exit code $?"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Check what the finished run wrote
docker exec jupyter python3 -c "import pandas as pd; r=pd.read_csv('report.csv'); print(r.shape, r['holiday'].notna().sum(), r['date'].is_unique, r['hires'].isna().sum())"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
A common mistake
Warning
Many people think that a pipeline which finished without an error is right. It only passed the checks somebody wrote. The row counts here say nothing about whether the weather words are spelt alike. Add a check for each fault you have seen.
The stage pattern worked out in Python
# The stage pattern in plain Python: do a step, report, stop if too few rows are left.
rows = [3, 8, 1, 9, 4, 2]
least = 3
def keep_over(rows, limit):
kept = []
for x in rows:
if x > limit:
kept.append(x)
return kept
def stage(name, rows, limit):
rows = keep_over(rows, limit)
print(name, "->", len(rows), "rows")
if len(rows) < least:
print("STOP:", name)
return None
return rows
rows = stage("keep over 2", rows, 2)
if rows is not None:
rows = stage("keep over 5", rows, 5)
if rows is None:
print("stopped")
else:
print("finished")
# Change least to 2 and run again.
Going further: add a check
Add a stage to steps/pipeline.py that stops if any weather word is not one of cloudy, rainy or sunny. Use set(frame["weather"]) and compare.
Try it on the cleaned data, then on the data before the strip line. What does the check catch that the row counts missed?