Data Pipelines and Orchestration

From Scheduled Script to Orchestrated Pipeline#

A scraper that runs on cron and writes straight to a database works until it grows five stages, a stage fails at 2am with no trace, or a retry double-writes rows. Orchestration is the discipline of running multi-stage data jobs as a graph of dependent tasks rather than one stubborn script: each stage declares its inputs, outputs, retry policy and schedule, and the system decides what runs next and records what happened.

Why Cron Stops Being Enough#

Cron runs a command on a clock and forgets. It cannot express "run load only after fetch succeeded, rerun load if it died midway, but never repeat fetch". It has no idea yesterday's run failed, so it cannot catch up. It will happily start a second copy while the first is still running. Once a pipeline has more than one stage, you need four things cron lacks:

• a recorded dependency graph, so stages run in the right order;
• per-task retries with backoff, so a flaky fetch does not rerun the whole job;
• run history, so you can see what ran, when, with which inputs and how it ended;
• parameterised runs, so you can re-run "the job for 14 September" without editing code.

Orchestrators such as Prefect, Airflow and Dagster provide these. So can a disciplined hand-rolled runner, which is worth building once to understand what the frameworks do for you.

Rule 1: Every Task Is Idempotent#

Idempotent means running a task twice with the same input leaves the world in the same state as running it once. It is what makes retries safe, and without it no orchestrator can help you. The canonical scraping flow has three stages, each with its own idempotency rule:

• fetch is resumable: it skips URLs already cached for this run (see caching and resume).
• clean is a pure function: it reads its input and writes a new output, never mutating the input.
• load upserts by a stable natural key, never blindly appends.

The load step is where most double-write bugs live. In SQLite (and similarly in PostgreSQL) an upsert looks like this:

import sqlite3

def load(rows, db_path="catalog.db"):
    con = sqlite3.connect(db_path)
    con.execute('''CREATE TABLE IF NOT EXISTS products (
        sku TEXT PRIMARY KEY, title TEXT, price REAL, updated_at TEXT)''')
    con.executemany('''
        INSERT INTO products (sku, title, price, updated_at)
        VALUES (:sku, :title, :price, :updated_at)
        ON CONFLICT(sku) DO UPDATE SET
            title = excluded.title,
            price = excluded.price,
            updated_at = excluded.updated_at
        WHERE excluded.price IS NOT products.price
           OR excluded.title IS NOT products.title''', rows)
    con.commit()
    con.close()

Run load three times with the same rows and the table still has one row per SKU. The WHERE clause means updated_at only moves when a value actually changed, which keeps freshness queries honest.

Rule 2: Stage Outputs Are Persisted Artifacts#

A subtle bug: load reads clean's result from an in-memory list. The moment a stage is retried in a fresh process, that data is gone and the stage either crashes or, worse, loads nothing and reports success. Make every stage write its output to a persisted artifact (a file, a table, an object-store key) and pass the location, not the data:

import json
from pathlib import Path

ROOT = Path("runs")

def artifact(run_id: str, stage: str, name: str) -> Path:
    p = ROOT / run_id / stage
    p.mkdir(parents=True, exist_ok=True)
    return p / name

def write_jsonl(path: Path, rows) -> int:
    tmp = path.with_suffix(".tmp")
    n = 0
    with open(tmp, "w", encoding="utf-8") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
            n += 1
    tmp.replace(path)  # atomic: readers never see half a file
    return n

def read_jsonl(path: Path):
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f]

The artifacts make the pipeline inspectable: "what did fetch return for the run that broke load?" becomes ls runs/2026-09-14/fetch/ plus head items.jsonl, not a search through logs. They also let you re-run clean against yesterday's fetch after fixing a parser bug, without touching the network.

A Minimal Hand-Rolled Runner#

Before adopting a framework, it helps to see how little code the core ideas take. This runner executes stages in order, skips stages that already completed for the run, retries with exponential backoff, and records everything in a status table:

import logging
import sqlite3
import time
from datetime import datetime, timezone

log = logging.getLogger("pipeline")

def now():
    return datetime.now(timezone.utc).isoformat()

def run_pipeline(run_id, stages, db="pipeline.db", retries=3):
    con = sqlite3.connect(db)
    con.execute('''CREATE TABLE IF NOT EXISTS stage_runs (
        run_id TEXT, stage TEXT, status TEXT, items INTEGER,
        error TEXT, started TEXT, finished TEXT,
        PRIMARY KEY (run_id, stage))''')
    for name, fn in stages:
        done = con.execute(
            "SELECT 1 FROM stage_runs WHERE run_id=? AND stage=? AND status='ok'",
            (run_id, name)).fetchone()
        if done:
            log.info("run=%s stage=%s skipped (already ok)", run_id, name)
            continue
        for attempt in range(1, retries + 1):
            started = now()
            try:
                items = fn(run_id)
                con.execute("INSERT OR REPLACE INTO stage_runs VALUES (?,?,?,?,?,?,?)",
                            (run_id, name, "ok", items, None, started, now()))
                con.commit()
                log.info("run=%s stage=%s ok items=%s", run_id, name, items)
                break
            except Exception as exc:
                con.execute("INSERT OR REPLACE INTO stage_runs VALUES (?,?,?,?,?,?,?)",
                            (run_id, name, "failed", None, repr(exc), started, now()))
                con.commit()
                log.warning("run=%s stage=%s attempt=%d failed: %r",
                            run_id, name, attempt, exc)
                if attempt == retries:
                    raise  # downstream stages must not run on missing input
                time.sleep(30 * 2 ** (attempt - 1))

Each stage function takes the run_id, reads its input artifact, writes its output artifact and returns an item count. Calling run_pipeline("2026-09-14", [("fetch", fetch), ("clean", clean), ("load", load_stage)]) a second time after a crash in clean skips fetch entirely and resumes at clean. That is the behaviour cron cannot give you.

The Same Pipeline in Prefect#

A workflow framework takes the same functions and adds scheduling, a UI, run history and remote workers. In Prefect, tasks and flows are decorated Python functions:

from prefect import flow, task

@task(retries=3, retry_delay_seconds=[30, 60, 120])
def fetch(run_id: str) -> str:
    ...  # crawl, write runs/<run_id>/fetch/items.jsonl
    return f"runs/{run_id}/fetch/items.jsonl"

@task
def clean(raw_path: str) -> str:
    ...
    return raw_path.replace("/fetch/", "/clean/")

@task(retries=2)
def load(clean_path: str) -> int:
    ...

@flow(log_prints=True)
def nightly_crawl(run_date: str):
    load(clean(fetch(run_date)))

if __name__ == "__main__":
    nightly_crawl("2026-09-14")

Notice the tasks pass paths, not record lists. Orchestrators can pass return values between tasks, but large payloads bloat their result storage and metadata database; Airflow, for instance, intends XComs for small values. Persisted artifacts plus small pointers scale; lists of a million records in task results do not.

Choosing a Tool#

A common production shape combines two layers: an orchestrator schedules the daily graph, and inside the fetch task a queue of per-URL jobs is drained by a worker pool (see distributed crawling).

Backfills and Logical Dates#

Backfill means running the graph for past dates you missed. It only works if tasks take the date as a parameter instead of calling datetime.now() inside. A task that computes "today" internally will, when backfilled for 3 September, quietly fetch and label today's data as 3 September. Pass the logical date in, partition artifacts by it, and backfilling is a loop:

from datetime import date, timedelta

d = date(2026, 9, 1)
while d <= date(2026, 9, 7):
    run_pipeline(d.isoformat(), STAGES)
    d += timedelta(days=1)

Be careful what a backfill can actually recover. You can re-clean and re-load old raw artifacts at any time, but you cannot re-fetch the web as it was last Tuesday; prices and pages have moved on. That asymmetry is a strong reason to keep raw fetch output, covered in dataset versioning and lineage.

Failure Modes to Test Deliberately#

• Kill the process mid-load. Rerun. Row counts must match a clean run exactly.
• Make fetch return zero items. The pipeline should fail or alert, not load an empty file and mark the run green. Add a minimum-count check between stages.
• Start two runs of the same date at once. Only one should proceed; use a lock or the orchestrator's concurrency limit.
• Change the parser and re-run clean on last week's artifacts. It should work without touching the network.

Finally, put the run_id on every log line and keep the status table queryable. The scheduling and monitoring lesson turns that table into staleness and volume-collapse alerts, so the pipeline explains itself when it wakes you at 4am. Frameworks are optional; idempotent tasks, persisted artifacts and recorded state are not.