M62 · Pipelines, Orchestration & Data Quality

Pipeline design principles

Every principle below exists because something will go wrong: the job will crash halfway, the source will resend a file, a bug will be found in last month's logic, the network will blip. Design for re-runs from day one.

Idempotency

Running a job twice with the same input leaves the same result as running it once. Achieved with upserts on natural keys, or by atomically replacing a whole partition ("delete day X, insert day X" in one transaction). Never with blind appends.

Determinism

The same input always gives the same output. Avoid datetime.now(), unseeded randomness, and unordered iteration inside transforms; pass the logical run date in as a parameter instead.

Incremental loads

Process only new or changed data, tracked with a watermark (the highest updated_at or id seen) or a manifest of processed files. Full reloads stay available as a fallback.

Backfills

Re-running a pipeline for a historical range (after fixing a bug or adding a column). Easy only if runs are parameterized by date and idempotent.

Retries

Transient failures (timeouts, rate limits, locks) are retried with exponential backoff; permanent failures (bad credentials, schema errors) fail fast and alert.

Atomicity

Outputs appear all at once or not at all: write to a temp file and rename, or load in one database transaction. Readers never see half a batch.

See idempotency in the visualizer

Both visualizers process the same twelve source records. Run each once, then press Re-run. The first loads with a plain INSERT (append), the second with an upsert keyed on the record id. Watch the target row counts after the second run.

Delivery semantics

When data moves between systems, a failure can happen after work is done but before it is acknowledged. That forces a choice:

Guarantee	How it happens	Risk
At-most-once	Acknowledge (or commit the offset) before processing	A crash loses data
At-least-once	Process, then acknowledge; on crash, redo	Duplicates after a crash
Exactly-once	Processing and acknowledgement commit atomically (transactions), or at-least-once plus an idempotent sink	Complexity and cost

In practice, most systems choose at-least-once delivery with idempotent writes: duplicates may be delivered, but the upsert absorbs them, so the result is as if each record were processed exactly once ("effectively once").

Remember: assume every step will run more than once; make writes idempotent and outputs atomic.

Incremental loads, watermarks, and backfills

A full reload copies the whole source every run. That's simple and correct, but it gets slower every day. An incremental load copies only rows that changed since the last run. The pipeline stores a watermark, the highest change marker it has processed (an updated_at timestamp or an auto-increment id), and next time asks the source for rows above it.

sequenceDiagram
  participant P as Pipeline
  participant S as State store
  participant Src as Source DB
  participant W as Warehouse
  P->>S: read watermark (2026-03-05 10:00)
  P->>Src: SELECT ... WHERE updated_at > watermark
  Src-->>P: 42 changed rows
  P->>W: upsert 42 rows (one transaction)
  P->>S: watermark = max(updated_at) of the batch

import sqlite3

src = sqlite3.connect(":memory:")
src.execute("CREATE TABLE customers (id INTEGER PRIMARY KEY, email TEXT, updated_at TEXT)")
src.executemany("INSERT INTO customers VALUES (?, ?, ?)", [
    (1, "ada@x.io", "2026-03-01 09:00"), (2, "ben@x.io", "2026-03-01 12:00"), (3, "cy@x.io", "2026-03-02 08:00")])

wh = sqlite3.connect(":memory:")
wh.execute("CREATE TABLE dim_customer (id INTEGER PRIMARY KEY, email TEXT, updated_at TEXT)")
wh.execute("CREATE TABLE etl_state (key TEXT PRIMARY KEY, value TEXT)")

def run():
    row = wh.execute("SELECT value FROM etl_state WHERE key = 'customers'").fetchone()
    watermark = row[0] if row else ""                      # "" sorts before any ISO timestamp
    changed = src.execute("SELECT id, email, updated_at FROM customers WHERE updated_at > ? ORDER BY updated_at",
                          (watermark,)).fetchall()
    with wh:                                               # one transaction: data and watermark together
        wh.executemany("""INSERT INTO dim_customer VALUES (?, ?, ?)
                          ON CONFLICT (id) DO UPDATE SET email = excluded.email, updated_at = excluded.updated_at""",
                       changed)
        if changed:
            wh.execute("INSERT INTO etl_state VALUES ('customers', ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value",
                       (changed[-1][2],))
    print(f"watermark {watermark or '(none)':16} -> loaded {len(changed)} rows")

run()                                                      # first run: everything
run()                                                      # nothing changed: 0 rows
src.execute("UPDATE customers SET email = 'ben@new.io', updated_at = '2026-03-03 10:00' WHERE id = 2")
src.execute("INSERT INTO customers VALUES (4, 'di@x.io', '2026-03-03 11:00')")
run()                                                      # only the two changes
print(wh.execute("SELECT * FROM dim_customer ORDER BY id").fetchall())

Backfills

A backfill re-runs the pipeline for past periods. If each run is a pure function of its logical date (run(day) processes exactly that day's partition and replaces it atomically), a backfill is just a loop, and it can even run in parallel:

from datetime import date, timedelta

def backfill(start: date, end: date) -> None:
    day = start
    while day <= end:
        run_for_day(day)          # idempotent: deletes + reinserts that day's partition in one transaction
        day += timedelta(days=1)

Remember: watermark + upsert + same-transaction state = safe incremental loads; date-parameterized, idempotent runs make backfills trivial.

Build a mini pipeline framework

Orchestrators look complicated, but their core is small: tasks are functions, dependencies form a directed acyclic graph (DAG), a scheduler runs tasks in dependency order, failures are retried, and state is recorded so a re-run can resume. You'll build each piece with the standard library.

Tasks and a DAG with graphlib

graphlib.TopologicalSorter (stdlib since 3.9) takes a mapping node → predecessors. static_order() gives one valid order; the prepare() / get_ready() / done() protocol tells you which tasks are ready right now, which is how a parallel scheduler works. Cycles raise graphlib.CycleError.

from graphlib import TopologicalSorter

TASKS = {}

def task(*deps):
    """Register a function as a task that depends on the named tasks."""
    def register(fn):
        TASKS[fn.__name__] = (fn, deps)
        return fn
    return register

@task()
def extract(results):
    return [{"id": 1, "amount": "10.5"}, {"id": 2, "amount": "oops"}, {"id": 3, "amount": "7"}]

@task("extract")
def validate(results):
    return [r for r in results["extract"] if r["amount"].replace(".", "", 1).isdigit()]

@task("validate")
def transform(results):
    return [{**r, "amount": float(r["amount"])} for r in results["validate"]]

@task("transform")
def load(results):
    return f"loaded {len(results['transform'])} rows"

@task("validate")
def report(results):
    return f"rejected {len(results['extract']) - len(results['validate'])} rows"

graph = {name: deps for name, (_, deps) in TASKS.items()}
sorter = TopologicalSorter(graph)
sorter.prepare()
results, wave = {}, 1
while sorter.is_active():
    ready = sorted(sorter.get_ready())          # these could run in parallel
    print(f"wave {wave}: {ready}")
    for name in ready:
        results[name] = TASKS[name][0](results)
        sorter.done(name)
    wave += 1
print(results["load"], "|", results["report"])

Retries with exponential backoff

A retry decorator wraps a task so transient errors are retried after a growing delay (1 s, 2 s, 4 s…), with jitter (a random fraction) so a thousand workers don't retry in lockstep. Only retry errors that can succeed later.

import functools, random, time

def retry(times=3, exceptions=(ConnectionError, TimeoutError), base_delay=0.01, jitter=0.5):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 2):
                try:
                    return fn(*args, **kwargs)
                except exceptions as exc:
                    if attempt > times:
                        raise
                    delay = base_delay * 2 ** (attempt - 1) * (1 + random.random() * jitter)
                    print(f"  attempt {attempt} failed ({exc!r}); retrying in {delay*1000:.0f} ms")
                    time.sleep(delay)
        return wrapper
    return decorator

calls = 0

@retry(times=3)
def fetch_page():
    global calls
    calls += 1
    if calls < 3:
        raise ConnectionError("connection reset")
    return "<html>ok</html>"

print(fetch_page(), "after", calls, "calls")

@retry(times=3)
def bad_credentials():
    raise PermissionError("401 Unauthorized")      # not retryable: fails immediately

try:
    bad_credentials()
except PermissionError as exc:
    print("failed fast:", exc)

State and checkpoints

A checkpoint records progress so a crashed run can resume instead of starting over. Write state atomically: write a temporary file, then Path.replace() it over the old one. A rename is atomic on the same filesystem, so a crash never leaves a half-written state file.

Remember: DAG + topological order + retries for transient errors + atomic checkpoints is the heart of every orchestrator.

Orchestration tools

An orchestrator schedules and runs pipelines and gives you what your mini framework lacks: a UI, history, alerting, distributed workers, backfill commands, secrets, and sensors that wait for data. The main options:

Tool	Model	Strengths	Watch out for
cron	Time-based schedule for commands	Everywhere, zero setup	No dependencies, retries, history, or alerting; overlapping runs
Apache Airflow	DAGs of tasks, defined in Python, scheduled by logical date	Huge ecosystem of operators, mature, the industry default	Heavier to run; passing data between tasks (XCom) is for small values
Prefect	Plain Python functions as `@flow`/`@task`	Minimal ceremony, dynamic workflows, good local dev	Fewer prebuilt integrations than Airflow
Dagster	Software-defined assets: you declare the tables you want, it derives the graph	Lineage, partitions, data-aware scheduling, type checks	A different mental model (assets, not tasks)

# m h dom mon dow   command  (02:30 every day; flock prevents overlapping runs)
30 2 * * *  cd /srv/etl && flock -n /tmp/etl.lock pixi run python -m etl.pipeline raw wh.db state >> etl.log 2>&1

Remember: cron schedules commands; orchestrators add dependencies, retries, history, backfills, and visibility. Airflow is task-centric, Prefect is function-centric, Dagster is asset-centric.

Data quality

Bad data is worse than no data: it produces confident wrong answers. Quality checks belong at the pipeline's front door (reject or quarantine bad input) and at its exit (verify outputs before publishing). The checks you will write again and again:

Schema: expected columns exist with expected types.
Completeness: required fields are not null; row counts are within the normal range.
Uniqueness: keys don't repeat.
Validity: values in range (0 <= fare < 500) or in an allowed set; formats match.
Consistency: end_time > start_time; totals match line items.
Referential integrity: foreign keys point at existing dimension rows.
Freshness: the newest record is recent enough.

Decide per rule what a failure does: reject the row (to a rejected-rows report or quarantine table with a reason), warn (log and continue), or fail the run (stop before publishing, for problems such as a missing column or a 90% drop in volume).

from datetime import datetime

rows = [
    {"trip_id": "T1", "start": "2026-03-05 08:00", "end": "2026-03-05 08:20", "station": "S01", "fare": "3.50"},
    {"trip_id": "T2", "start": "2026-03-05 09:00", "end": "2026-03-05 08:00", "station": "S02", "fare": "2.00"},
    {"trip_id": "T3", "start": "not a date",       "end": "2026-03-05 10:00", "station": "S99", "fare": ""},
    {"trip_id": "T1", "start": "2026-03-05 08:00", "end": "2026-03-05 08:20", "station": "S01", "fare": "3.50"},
    {"trip_id": "T4", "start": "2026-03-05 11:00", "end": "2026-03-05 11:30", "station": "S02", "fare": "950"},
]
STATIONS = {"S01", "S02"}

def parse(ts):
    try:
        return datetime.fromisoformat(ts)
    except ValueError:
        return None

def rule_required(r):
    return [f"missing:{k}" for k, v in r.items() if not v]

def rule_times(r):
    s, e = parse(r["start"]), parse(r["end"])
    if s is None or e is None:
        return ["bad_timestamp"]
    return ["end_before_start"] if e <= s else []

def rule_station(r):
    return [] if r["station"] in STATIONS else ["unknown_station"]

def rule_fare(r):
    return [] if not r["fare"] or 0 <= float(r["fare"]) < 500 else ["fare_out_of_range"]

RULES = [rule_required, rule_times, rule_station, rule_fare]
valid, rejected, seen = [], [], set()
for r in rows:
    reasons = [reason for rule in RULES for reason in rule(r)]
    if r["trip_id"] in seen:
        reasons.append("duplicate_trip_id")
    if reasons:
        rejected.append((r["trip_id"], reasons))
    else:
        valid.append(r)
        seen.add(r["trip_id"])
print("valid:", [r["trip_id"] for r in valid])
for trip_id, reasons in rejected:
    print("rejected", trip_id, reasons)

With pandas, column-level checks are vectorized one-liners, which is how you'd profile a whole batch:

Libraries

You rarely hand-roll quality checks at scale. pandera declares DataFrame schemas with typed columns and checks (pandas and Polars). Great Expectations (GX) manages suites of "expectations" with HTML data docs. pydantic validates individual records (API payloads, JSON events) into typed models. dbt has built-in unique/not_null/accepted_values/relationships tests, and Soda offers YAML checks.

Remember: check schema, completeness, uniqueness, validity, consistency, references, and freshness; every rejection carries a reason; decide per rule whether to reject, warn, or fail.

Observability: logs, metrics, lineage

When a pipeline misbehaves at 3 a.m., you need to answer: did it run, what did it read, what did it write, how long did it take, and what changed? Three kinds of signals answer those questions.

Structured logs: events as key/value records (JSON), not prose. A log aggregator can then answer "show every run where rows_rejected > 100". Production code uses structlog instead of print(), with a run_id bound to every line.
Metrics: numbers tracked over time (rows in/out, rejection rate, duration, freshness lag), with alerts on anomalies, for example "row count 60% below the 7-day average".
Lineage: which datasets each job read and wrote, so you can trace a bad dashboard number back to its source and see what a broken table affects downstream. OpenLineage is the open standard; Airflow, Dagster, Spark, and dbt can emit it.

import json, uuid
from datetime import datetime, UTC

def run_record(job, inputs, outputs, metrics):
    return {
        "eventType": "COMPLETE",
        "eventTime": datetime(2026, 3, 6, 2, 31, tzinfo=UTC).isoformat(),
        "run": {"runId": str(uuid.uuid5(uuid.NAMESPACE_URL, f"{job}/2026-03-05"))},   # deterministic for the demo
        "job": {"namespace": "spokes", "name": job},
        "inputs": [{"namespace": "file", "name": n} for n in inputs],
        "outputs": [{"namespace": "sqlite", "name": n} for n in outputs],
        "metrics": metrics,
    }

rec = run_record("bike_trips.load", ["raw/trips_2026-03-05.csv", "raw/stations_2026-03-05.csv"],
                 ["warehouse.fact_trip", "warehouse.dim_station"],
                 {"rows_read": 34, "rows_rejected": 3, "rows_loaded": 30, "rejection_rate": round(3 / 34, 3)})
print(json.dumps(rec, indent=2))

history = [228, 231, 225, 240, 236, 229, 233]           # rows loaded on previous days
today = 92
mean = sum(history) / len(history)
if today < 0.5 * mean:
    print(f"ALERT: loaded {today} rows, {today / mean:.0%} of the 7-day mean ({mean:.0f})")

Remember: log events with structured fields and a run id, track volume and freshness metrics with alerts, and record lineage so you can trace and assess impact.

Testing pipelines

Pipelines are code and deserve the same test pyramid as any other code, plus a few data-specific tests:

Unit tests for pure transforms: given these input rows, expect these output rows. Fast and numerous. Keep transforms pure (no I/O) precisely so they are easy to test.
Integration tests run the pipeline end to end against real files in tmp_path and a real (temporary) database.
Idempotency tests: run twice, assert identical row counts and contents.
Incremental tests: deliver data in two batches; the result must equal one full load.
Edge-case fixtures: duplicates, nulls, late data, time-zone boundaries, empty inputs.
Data tests in production: the quality checks above, run on every batch.

# tests/integration/test_pipeline.py      (pixi run pytest)
import sqlite3
import pytest
from etl.generator import generate
from etl.pipeline import run_pipeline

@pytest.mark.integration
def test_running_twice_is_idempotent(tmp_path):
    generate(tmp_path / "raw")
    first = run_pipeline(tmp_path / "raw", tmp_path / "wh.db", tmp_path / "state")
    (tmp_path / "state" / "manifest.json").unlink()            # force a full re-read
    second = run_pipeline(tmp_path / "raw", tmp_path / "wh.db", tmp_path / "state")
    assert second.fact_trip_rows == first.fact_trip_rows

@pytest.mark.integration
def test_two_batches_equal_one_full_load(tmp_path):
    ...  # deliver day 1, run; deliver day 2, run; compare with a fresh full run

Remember: pure transforms get unit tests; the whole pipeline gets idempotency and incremental tests on temporary files and databases.

Practice

Write retry(times, exceptions), a decorator factory. The wrapped function is called up to times + 1 times. If it raises one of exceptions (a tuple), call it again; if the last attempt fails, re-raise that exception. Any other exception propagates immediately. Also record the number of calls made in the most recent invocation as the attribute wrapper.attempts. Preserve the function's name with functools.wraps. No sleeping is needed.

Hint

Inside wrapper, loop for attempt in range(1, times + 2), set wrapper.attempts = attempt before each call, and re-raise with a bare raise when attempt == times + 1.

Show solution

import functools

def retry(times, exceptions):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 2):
                wrapper.attempts = attempt
                try:
                    return fn(*args, **kwargs)
                except exceptions:
                    if attempt == times + 1:
                        raise
        wrapper.attempts = 0
        return wrapper
    return decorator

Write waves(deps) where deps maps each task to a list of tasks it depends on. Return a list of lists: each inner list holds the tasks that can run in parallel at that step (all their dependencies are in earlier waves), sorted alphabetically. Use graphlib.TopologicalSorter with prepare(), get_ready(), and done(). A cycle should raise graphlib.CycleError.

Hint

prepare() raises CycleError for you. Then loop while ts.is_active(): take ready = sorted(ts.get_ready()), append it, and call ts.done(*ready).

Show solution

from graphlib import TopologicalSorter

def waves(deps):
    ts = TopologicalSorter(deps)
    ts.prepare()
    out = []
    while ts.is_active():
        ready = sorted(ts.get_ready())
        out.append(ready)
        ts.done(*ready)
    return out

Write incremental_load(conn, source_rows). source_rows is the full current source: a list of (id, value, updated_at) tuples (ISO strings). The warehouse conn has (or you create, if missing) target(id INTEGER PRIMARY KEY, value TEXT, updated_at TEXT) and state(key TEXT PRIMARY KEY, value TEXT). Read the watermark from state where key = 'target' (none on the first run), upsert only rows with updated_at strictly greater than the watermark, store the new watermark (the max updated_at loaded) in the same transaction, and return the number of rows loaded.

Hint

Filter in Python: [r for r in source_rows if watermark is None or r[2] > watermark]. Use with conn: around the upserts and the state upsert.

Show solution

def incremental_load(conn, source_rows):
    conn.execute("CREATE TABLE IF NOT EXISTS target (id INTEGER PRIMARY KEY, value TEXT, updated_at TEXT)")
    conn.execute("CREATE TABLE IF NOT EXISTS state (key TEXT PRIMARY KEY, value TEXT)")
    row = conn.execute("SELECT value FROM state WHERE key = 'target'").fetchone()
    watermark = row[0] if row else None
    changed = [r for r in source_rows if watermark is None or r[2] > watermark]
    with conn:
        conn.executemany("""
            INSERT INTO target VALUES (?, ?, ?)
            ON CONFLICT (id) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
        """, changed)
        if changed:
            conn.execute("""
                INSERT INTO state VALUES ('target', ?)
                ON CONFLICT (key) DO UPDATE SET value = excluded.value
            """, (max(r[2] for r in changed),))
    return len(changed)

Write validate(rows, schema). schema maps column → a dict of rules, any of: "required": True, "unique": True, "min": n, "max": n, "allowed": {...}. Return (valid_rows, rejected) where rejected is a list of (row_index, reasons) and reasons are strings formatted "<rule>:<column>", for example "required:fare", "min:fare", "unique:trip_id". Rules:

A value of None or "" fails required and skips the other rules for that column.
unique: a value already seen in an earlier valid row is a duplicate.
Check columns in schema order and rules in the order required, unique, min, max, allowed.

Hint

Collect reasons for one row into a list. Only add the row's unique values to the seen sets if the row ends up valid.

Show solution

def validate(rows, schema):
    valid, rejected = [], []
    seen = {col: set() for col, rules in schema.items() if rules.get("unique")}
    for i, row in enumerate(rows):
        reasons = []
        for col, rules in schema.items():
            value = row.get(col)
            if value is None or value == "":
                if rules.get("required"):
                    reasons.append(f"required:{col}")
                continue
            if rules.get("unique") and value in seen[col]:
                reasons.append(f"unique:{col}")
            if "min" in rules and value < rules["min"]:
                reasons.append(f"min:{col}")
            if "max" in rules and value > rules["max"]:
                reasons.append(f"max:{col}")
            if "allowed" in rules and value not in rules["allowed"]:
                reasons.append(f"allowed:{col}")
        if reasons:
            rejected.append((i, reasons))
        else:
            valid.append(row)
            for col in seen:
                seen[col].add(row[col])
    return valid, rejected

Write run_steps(steps, state_path). steps is a list of (name, fn) pairs; state_path is a pathlib.Path to a JSON file shaped {"done": [names]} (missing on the first run). Skip steps already done; run the others in order, and after each success append its name and save the file atomically (write a .tmp sibling, then replace). If a step raises, let the exception propagate (earlier progress stays saved). Return the list of names actually run.

Hint

Load state with json.loads(state_path.read_text()) if state_path.exists() else {"done": []}. Save with tmp = state_path.with_suffix(".tmp"), tmp.write_text(...), tmp.replace(state_path).

Show solution

import json
from pathlib import Path

def run_steps(steps, state_path):
    state = json.loads(state_path.read_text()) if state_path.exists() else {"done": []}
    ran = []
    for name, fn in steps:
        if name in state["done"]:
            continue
        fn()
        state["done"].append(name)
        tmp = state_path.with_suffix(".tmp")
        tmp.write_text(json.dumps(state))
        tmp.replace(state_path)
        ran.append(name)
    return ran

Mini-project: an incremental ETL with a validation report

Simulate an e-commerce source database (SQLite) where orders are inserted and updated over time, and build a pipeline that loads them incrementally into a warehouse database, validating every batch.

pixi init orders-etl && cd orders-etl
pixi add python structlog pytest
pixi run python simulate.py --day 2026-03-01    # inserts/updates orders in source.db with a fixed seed
pixi run python -m orders_etl                   # one incremental run
pixi run pytest

simulate.py adds about 100 orders per day and updates about 10 existing ones (status changes), setting updated_at; include some bad rows (negative totals, unknown customer ids, null emails).
The pipeline is a DAG built on graphlib: extract (rows above the watermark, with a 5-minute lookback), validate, load (upsert), report.
extract is wrapped in your retry decorator; inject a random ConnectionError to prove it.
validate applies at least six rules and writes reports/run-<run_id>.json with counts by reason and the rejected rows.
Data and watermark commit in one transaction; every step logs JSON with structlog and a bound run_id.

Running the pipeline twice in a row loads 0 rows the second time
Three daily runs produce the same warehouse as one run after three days (tested)
Killing the process mid-run (raise in load) leaves the watermark unchanged, and the next run recovers
The validation report lists every rejected row with at least one reason
The run fails (non-zero exit) if more than 20% of a batch is rejected

Stretch: turn the DAG into a Prefect flow or Dagster assets; add a --backfill START END option that re-processes a date range idempotently; emit an OpenLineage-shaped JSON event per run.

Quiz

Recap

Design every pipeline to be idempotent, deterministic, atomic, incremental, and safe to retry and backfill.
At-least-once delivery plus idempotent writes gives effectively-once results.
Watermarks drive incremental loads; store them with the data, use a lookback, and plan for deletes.
A DAG plus graphlib.TopologicalSorter, a retry decorator, and atomic checkpoints form a working mini orchestrator.
cron schedules commands; Airflow (tasks), Prefect (functions), and Dagster (assets) add dependencies, retries, history, and backfills.
Quality checks cover schema, completeness, uniqueness, validity, consistency, references, and freshness, and every rejection carries a reason.
Observe with structured logs, volume/freshness metrics with alerts, and lineage; test idempotency and incremental equivalence.

Going further

graphlib and functools.wraps in the Python docs
Apache Airflow, Prefect, and Dagster documentation
pandera, Great Expectations, and pydantic
structlog and OpenLineage
Maxime Beauchemin, "Functional Data Engineering: a modern paradigm for batch data processing"
Next: M63 · Streaming & Big Data, then Project 9