Training Pipelines and Orchestration

A training pipeline is a directed acyclic graph of steps with declared inputs and outputs, run by an orchestrator that supplies scheduling, retries, logs, and resources. It is the same DAG DVC describes, elevated from "run the script in order" to "run this step on this machine and retry exactly the part that failed."

The idea

An orchestrator contributes five things to a pipeline:

• A task graph — Airflow, Prefect, Dagster, Kubeflow, or Step Functions.
• An execution backend — local process, container, Kubernetes pod, or batch job.
• Retries and timeouts per task, with backoff.
• Observability — structured logs, run history, and alerting on failure.
• Resource requests — CPU, memory, and GPU per step, so a GPU step does not run on a CPU node by accident.

Four design rules separate pipelines that survive from pipelines that rot:

• Idempotent steps. Rerunning a task must not double-write an output. Write to a temp path and rename, or check for existing output first.
• Parameters, not constants. Secrets and buckets come from config, never from a hardcoded string.
• Artifacts by reference. Pass a URI plus hash, not a copied file.
• Cheapest disqualifier first. Validate data before spending GPU hours.

Worked example

A pipeline that consumes a new data drop has five steps: ingest, validate, train, evaluate, register.

The ordering is the design. validate checks schema, null rate, and label distribution and costs seconds. If the label column is missing, it fails and steps 3–5 never run, so no GPU time is spent. If instead train ran first and the failure surfaced at evaluate, you would have paid for a full run to learn something a schema check already knew.

The same reasoning applies to the gate: register is a task with a condition, not a manual step. It runs only when evaluate reports a metric that clears the threshold against the incumbent.

In code

from prefect import flow, task

@task(retries=2, retry_delay_seconds=30)
def validate(uri: str) -> str:
    report = check_schema(uri)  # columns, dtypes, null rate
    if report.null_rate > 0.05:
        raise ValueError(f"null rate {report.null_rate:.2%}")
    return uri

@task(retries=1, timeout_seconds=7200)
def train(uri: str) -> str:
    model = fit(load(uri))
    save(model, "s3://fanout-models/candidate/")
    return "s3://fanout-models/candidate/"

@flow(name="churn-train")
def pipeline(uri: str) -> str:
    good = validate(uri)
    artifact = train(good)
    score = evaluate(artifact)
    return register(artifact) if score >= 0.80 else "rejected by gate"

Raising inside validate marks the task failed, so the orchestrator never schedules train.

Check yourself

1. Why should data validation be a separate step before training rather than a check inside it?
2. What makes a pipeline step safe to retry after a mid-run crash?
3. How should a step hand a trained model to the next step?

Key takeaways

• Order steps cheapest-disqualifier-first to avoid wasting GPU time.
• Every step should be idempotent, parameterized, and explicit about its inputs.
• Orchestration adds retries, resources, and audit — it does not replace the DAG.