Orchestration, streaming and lineage

A pipeline that runs once is a script. Making it run every day, in the right order, and tell you exactly which step failed is a different job. This section covers choosing between batch and streaming honestly, since streaming sounds better and is considerably harder to operate; what an orchestrator gives you beyond a cron entry; what to do with events that arrive late or out of order; and lineage, which is being able to answer where a number came from when someone disputes it.

Batch or streaming#

Batch processes accumulated data on a schedule. Streaming processes each event as it arrives. Streaming sounds better and is much harder to operate. Choose it only when freshness genuinely matters to the decision the data feeds.

Batch, hourly or daily       reports, training data, most analytics
Streaming, seconds           fraud, alerting, live personalisation

Ask: would anyone act differently if this data were an hour old?
If not, batch.

The part that is easy to underestimate is the operating cost. A batch job that fails is a red square and a rerun in the morning. A streaming job that fails is consumer lag climbing while someone works out whether it is safe to restart from the last committed offset, and lag that grows faster than the job can catch up is an outage with no natural end. Exactly-once, where a framework offers it, holds inside that framework's own state and its own sinks. The moment the job calls an external API or writes to a database it does not manage, you are back to at-least-once and the write has to be idempotent yourself.

Orchestration#

Pipelines have dependencies: this table needs that one to finish first. An orchestrator runs steps in dependency order, retries failures, and records what ran. The value is less in running things and more in knowing, after a failure, exactly which steps completed. The core of one is smaller than the products suggest, and writing it once makes clear what you are actually buying.

def run_dag(tasks, deps, attempts=2):
    """tasks: name -> callable. deps: name -> names that must finish first."""
    state = {name: "pending" for name in tasks}

    def ready(name):
        return all(state[d] == "ok" for d in deps.get(name, ()))

    progressed = True
    while progressed:
        progressed = False
        for name, fn in tasks.items():
            if state[name] != "pending" or not ready(name):
                continue
            for attempt in range(attempts):
                try:
                    fn()
                    state[name] = "ok"
                    break
                except Exception as e:
                    state[name] = f"failed: {e}"
            progressed = True

    # Anything still pending was blocked by an upstream failure, not broken itself.
    for name in state:
        if state[name] == "pending":
            state[name] = "skipped: upstream failed"
    return state

calls, tried = [], {"clean": 0}

def extract():
    calls.append("extract")

def clean():
    tried["clean"] += 1
    if tried["clean"] == 1:
        raise RuntimeError("bad row")      # fails the first time, then works
    calls.append("clean")

def report():
    calls.append("report")

tasks = {"extract": extract, "clean": clean, "report": report}
deps = {"clean": ["extract"], "report": ["clean"]}
print(run_dag(tasks, deps))
# {'extract': 'ok', 'clean': 'ok', 'report': 'ok'}  -- clean failed once and retried
print(calls)                               # ['extract', 'clean', 'report']

What a real orchestrator adds on top is a scheduler that knows about dates, parallel execution with a concurrency cap, a record of every past run, and a page that shows which day of which task is red. Cron gives you none of that and is fine for one independent job. The moment a second job reads the first one's output, you need the dependency edge, because otherwise the second job runs on yesterday's data whenever the first one is late and nothing anywhere says so.

Late and out-of-order data#

Events arrive late, out of order, or twice. A phone was in a tunnel; a partner batches their exports; a retry duplicated a send. A pipeline that assumes ordered arrival produces aggregates that were right when computed and are wrong an hour later, which is the worst combination because nothing errors and the number moves after someone quoted it.

There are two clocks and keeping them apart is most of the battle. Event time is when the thing happened. Processing time is when your job saw it. Aggregate by event time or the Monday figure includes whatever straggled in on Tuesday. A watermark is your declared position in event time: the point past which you promise not to accept anything more.

from collections import defaultdict

def aggregate(events, allowed_lateness=2):
    """events: (event_time, processing_order) pairs, in arrival order.
    Counts per event-time bucket, with a watermark that closes old buckets."""
    counts, watermark, dropped = defaultdict(int), None, []
    for event_time, _ in events:
        watermark = event_time if watermark is None else max(watermark, event_time)
        if event_time < watermark - allowed_lateness:
            dropped.append(event_time)          # too late: count it separately
            continue
        counts[event_time] += 1
    return dict(counts), dropped

arrivals = [(10, 0), (11, 1), (9, 2), (12, 3), (8, 4), (12, 5)]
print(aggregate(arrivals))
# ({10: 1, 11: 1, 9: 1, 12: 2}, [8])  -- the event from time 8 arrived after the
# watermark reached 12, so it is dropped rather than silently changing a closed bucket

Dropping is one policy of three and you must choose explicitly. Drop, and the bucket is stable but slightly under-counts. Accept and restate, and the bucket is accurate but a number in a dashboard changes after someone screenshotted it. Send late events to a side output and correct in a nightly batch, which is what most systems that care about both end up doing. Say which you chose in the documentation the analysts read, because they will find the discrepancy before you do.

Lineage#

Lineage records which inputs produced which outputs. When a number in a dashboard is wrong, lineage lets you trace it to the source rather than guessing. For ML it answers the question every audit asks: exactly which data was this model trained on. It is cheap to record at write time and impossible to reconstruct afterwards, which is why it is always missing when it is first needed.

import hashlib, json, datetime as dt

def manifest(output_path, inputs, code_version, row_count):
    """Write this next to every output. It is the whole of lineage at one step."""
    return {
        "output": output_path,
        "written_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "code_version": code_version,          # git sha of the transform
        "inputs": [
            {"path": p, "rows": n, "content_hash": h} for p, n, h in inputs
        ],
        "rows": row_count,
    }

def digest(rows):
    return hashlib.sha256(json.dumps(rows, sort_keys=True).encode()).hexdigest()[:12]

raw = [{"id": 1, "total": 10}, {"id": 2, "total": 5}]
print(manifest(
    "warehouse/daily/date=2026-03-01",
    [("raw/orders/date=2026-03-01", len(raw), digest(raw))],
    code_version="a91f0c3",
    row_count=len(raw),
))