What a data pipeline actually does, with 60 lines of Python you can run

A data pipeline moves data from the place it is produced to the place it gets used, and changes its shape on the way. That is the whole idea. Most explanations stop at a diagram with three boxes labelled extract, transform, load, which tells you nothing about what the code looks like or why it breaks.

So here is a pipeline you can actually run. About 60 lines of Python, no framework, no cloud account. Then I will go through what it deliberately does not handle, because that gap is where most of the real work lives.

The input, which is messy on purpose

Save this as orders.csv. It has five rows and two of them are broken, which is roughly the ratio you get from a real export.

id,customer,amount_cents,placed_at A-1001,Ada@Example.com ,124900,03/04/2026 A-1002, Grace@example.com,8050,03/05/2026 A-1003,lin@example.com,not_a_number,03/05/2026 A-1004,hopper@example.com,45000,2026-03-06 A-1005,ken@example.com,99900,03/07/2026 

Row A-1003 has a non-numeric amount. Row A-1004 has a date in a different format from the rest. The first two rows have whitespace and inconsistent email casing. Nobody writes a CSV like this on purpose. You get it when three different systems have written to the same file over two years.

The pipeline

import csv import logging import sqlite3 from datetime import datetime, timezone from pathlib import Path logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") SCHEMA = """ CREATE TABLE IF NOT EXISTS orders ( order_id TEXT PRIMARY KEY, customer TEXT NOT NULL, amount_usd REAL NOT NULL, placed_at TEXT NOT NULL, loaded_at TEXT NOT NULL ) """ UPSERT = """ INSERT INTO orders VALUES (:order_id, :customer, :amount_usd, :placed_at, :loaded_at) ON CONFLICT(order_id) DO UPDATE SET customer = excluded.customer, amount_usd = excluded.amount_usd, placed_at = excluded.placed_at, loaded_at = excluded.loaded_at """ def extract(path): with open(path, newline="", encoding="utf-8") as fh: yield from csv.DictReader(fh) def transform(row): return { "order_id": row["id"].strip(), "customer": row["customer"].strip().lower(), "amount_usd": int(row["amount_cents"]) / 100, "placed_at": datetime.strptime(row["placed_at"], "%m/%d/%Y").date().isoformat(), "loaded_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), } def run(csv_path, db_path): conn = sqlite3.connect(db_path) conn.execute(SCHEMA) clean, rejected = [], [] for row in extract(csv_path): try: clean.append(transform(row)) except (KeyError, ValueError) as exc: rejected.append((row.get("id", "<no id>"), str(exc))) conn.executemany(UPSERT, clean) conn.commit() logging.info("loaded %d rows, rejected %d", len(clean), len(rejected)) for order_id, why in rejected: logging.warning("rejected %s: %s", order_id, why) conn.close() if __name__ == "__main__": run(Path("orders.csv"), Path("orders.db")) 

Run it:

INFO loaded 3 rows, rejected 2 WARNING rejected A-1003: invalid literal for int() with base 10: 'not_a_number' WARNING rejected A-1004: time data '2026-03-06' does not match format '%m/%d/%Y' 

Three rows landed. Two were rejected with the reason attached to the order id, so you can go and ask someone about A-1003 instead of guessing which row died.

Four things are happening in those 60 lines, and each one maps to a decision you will have to make again on a larger pipeline.

extract is a generator. It yields one row at a time instead of loading the file into memory. On a 4 GB export this is the difference between a job that runs and a job the kernel kills.

transform converts rather than repairs. Cents become dollars, the US date format becomes an ISO date, emails get lowercased and trimmed. It raises on anything it cannot convert instead of writing a zero or a null. A null that should have been 1249.00 will survive every test you write and surface three months later in a revenue report.

The rejection list is separate from the load. Bad rows do not stop good rows, and they do not disappear either. This single choice is the difference between a pipeline people trust and one people re-run by hand every morning.

The load is an upsert, keyed on order_id. That is what makes the job safe to run twice.

Run it twice, then count the rows

row count: 3 ('A-1001', 'ada@example.com', 1249.0, '2026-03-04') ('A-1002', 'grace@example.com', 80.5, '2026-03-05') ('A-1005', 'ken@example.com', 999.0, '2026-03-07') 

Still three rows after two runs. If UPSERT were a plain INSERT, the second run would either crash on the primary key or, with a table that has no key, quietly double your revenue.

This property has a name worth knowing: idempotency. Running the job twice produces the same end state as running it once. You want it because you will re-run jobs. The network drops, a credential expires, someone fixes the source file at 2am and asks you to load it again. A pipeline you cannot safely re-run becomes a pipeline somebody has to babysit.

What these 60 lines do not handle

This is the honest part. The script above is a pipeline, but it is the first ten percent of one.

Schema drift. Someone renames amount_cents to amount and every row rejects at once. The fix is to validate the header against an expected contract and fail the whole run with a clear message, rather than reporting 40,000 individual rejections.

Partial failure and retries. If the load dies halfway through a large batch, executemany inside one transaction rolls back and you are clean. Start writing in chunks and committing per chunk, which you will do for memory reasons, and you now need to track which chunks landed.

Incremental loads. This script reads the whole file every time. At real volume you need a watermark, usually the maximum placed_at or an updated timestamp from the source, so each run only picks up what is new. Then you need a way to force a full reload when the watermark logic itself is wrong.

Late-arriving and corrected data. An order placed on the 4th can arrive on the 9th with a different amount. The upsert handles the correction. Any report you already sent out does not.

Observability. Right now the only record that the job ran is a log line in a terminal. The minimum useful version is a run table: start time, end time, rows in, rows loaded, rows rejected. Without it you cannot answer whether today's numbers are low because sales were low or because the job failed.

Orchestration. Something has to decide when this runs, what runs after it, and who gets told when it fails. That is what Airflow, Dagster and Prefect exist for, and it is also why you should not reach for them on day one.

The three failures that actually show up

On the data platform work I do at Brickclay, three problems come up far more often than any argument about tooling.

The silent partial load. The job reports success, loads 80 percent of the rows, and nobody notices for a week. Caused by swallowing exceptions in a loop. Fixed by counting rows in and rows out and failing when the difference is not explained by your rejection list.

The timezone that was never a timezone. A source writes local timestamps with no offset, the pipeline stores them as UTC, and every daily total is wrong by the number of hours in the shift. This one is expensive because the data looks completely reasonable.

The pipeline nobody can re-run. Usually a plain insert, or a transform that reads the current date, so re-running yesterday's job produces different output than it did yesterday. Notice that loaded_at in the script above has exactly this smell. It records when the row was written, which is fine for an audit trail, but it means the output bytes differ between runs. Keep that kind of field out of anything you compare run to run.

If you are starting one of these from scratch, build the rejection list and the run table before you add a scheduler. The scheduler makes the job automatic. Those two make it trustable, and on every project I have worked on, trust was the part that took longer.

I work on data and BI projects at Brickclay. If you have hit a failure mode I did not list, put it in the comments, I collect these.