Orchestration Options: Bare Pipeline vs Airflow vs Prefect

The Pipeline class in this toolkit is an in-process orchestrator: it runs

your steps in one Python process, in dependency order, right now. That is exactly

what you want for development, CI, batch jobs that fit in memory, and embedding a

pipeline inside a larger service. It is *not* a scheduler, a distributed executor,

or a UI.

This guide compares three ways to run the same Step graph and shows how to move

between them with minimal rewrites.

Table of contents

1. Decision tree

2. Comparison at a glance

3. Option A: the bare Pipeline

4. Option B: Apache Airflow

5. Option C: Prefect

6. Keeping step logic portable

7. Migration checklist

Decision tree

Does the whole job fit in one process and finish in a single run?
|
+-- YES: Do you need scheduling, retries across machines, or a UI?
|        |
|        +-- NO  -> Option A: bare Pipeline (this toolkit). Stop here.
|        +-- YES -> Is your team already on a scheduler?
|                   |
|                   +-- Airflow shop  -> Option B
|                   +-- Greenfield / Pythonic / dynamic -> Option C (Prefect)
|
+-- NO (data too big / steps need different machines / long-running):
         Use a distributed engine (Spark, Ray, Dask) for the heavy step and
         orchestrate the stages with Airflow or Prefect. The bare Pipeline can
         still run *inside* a single task.

The honest default for most tabular problems is Option A. Reach for B or C

only when you actually need scheduling, cross-process retries, backfills, or a

control-plane UI -- not before.

Comparison at a glance

Option A: the bare Pipeline

No extra infrastructure -- import, build, run.

from ml_pipeline import build_standard_pipeline, load_config

pipeline = build_standard_pipeline(load_config("configs/pipeline.yaml"))
run = pipeline.run()                 # executes ingest -> preprocess -> train -> evaluate
if not run.succeeded:
    raise SystemExit(run.report())
print(run["evaluate"].to_markdown())

Run it on a schedule with anything that can launch a process -- cron, a CI job, a

systemd timer, a Kubernetes CronJob:

# Retrain every night at 02:00 and write the evaluation report.
0 2 * * *  cd /opt/ml && /opt/ml/.venv/bin/python examples/run_pipeline.py --config configs/pipeline.yaml

Use when: the job fits in memory, runs start-to-finish, and you do not need a

scheduler/UI. Outgrow it when: you need cross-machine retries, backfills, SLAs,

or a control plane.

Option B: Apache Airflow

Airflow is a mature, battle-tested scheduler. Map each Step to an Airflow task

so Airflow owns scheduling, retries, and observability while your step logic stays

unchanged.

# dags/ml_pipeline_dag.py
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator

from ml_pipeline import load_config
from ml_pipeline.steps import (
    IngestionStep, PreprocessingStep, TrainingStep, EvaluationStep,
)

CONFIG = load_config("/opt/airflow/configs/pipeline.yaml")

def run_ingest(**ctx):
    ds = IngestionStep("ingest", **CONFIG["ingestion"]).run({})
    ctx["ti"].xcom_push(key="dataset", value=ds)   # small data only; see note

with DAG(
    "ml_pipeline",
    start_date=datetime(2024, 1, 1),
    schedule="0 2 * * *",
    catchup=False,
    default_args={"retries": 2},
) as dag:
    ingest = PythonOperator(task_id="ingest", python_callable=run_ingest)
    # ... define preprocess / train / evaluate similarly ...
    ingest >> preprocess >> train >> evaluate

XCom note: Airflow passes data between tasks via XCom, which is sized for

small metadata, not DataFrames or model objects. In practice each task should

persist its real output to shared storage (S3/GCS/a database) and push only the

*path* through XCom. The Dataset / PreprocessedData / TrainedModel objects

make natural artifacts to serialize at task boundaries.

Strengths: robust scheduler, backfills, mature UI, huge operator ecosystem.

Costs: you run a scheduler, a metadata DB, and workers; dynamic graphs are

more awkward than in plain Python.

Option C: Prefect

Prefect feels like decorating normal Python, which pairs naturally with this

toolkit's plain-callable steps.

# flows/ml_pipeline_flow.py
from prefect import flow, task

from ml_pipeline import load_config
from ml_pipeline.steps import (
    IngestionStep, PreprocessingStep, TrainingStep, EvaluationStep,
)

CONFIG = load_config("configs/pipeline.yaml")

@task(retries=2, retry_delay_seconds=30)
def ingest():
    return IngestionStep("ingest", **CONFIG["ingestion"]).run({})

@task
def preprocess(dataset):
    return PreprocessingStep("preprocess", dataset_step="ingest",
                             **CONFIG["preprocessing"]).run({"ingest": dataset})

@task
def train(pre):
    return TrainingStep("train", features_step="preprocess").run({"preprocess": pre})

@task
def evaluate(model, pre):
    return EvaluationStep("evaluate", model_step="train",
                          features_step="preprocess").run({"train": model, "preprocess": pre})

@flow(name="ml-pipeline")
def ml_pipeline():
    ds = ingest()
    pre = preprocess(ds)
    model = train(pre)
    return evaluate(model, pre)

if __name__ == "__main__":
    ml_pipeline()

Because the flow is ordinary Python, Prefect infers the dependency graph from how

the task results are passed -- and the data is passed in memory between tasks, so

there is no XCom-style size limit when running in a single environment.

Strengths: Pythonic, first-class dynamic graphs, schedules and event triggers,

gentle local-to-cloud path. Costs: a newer ecosystem; distributed runs still

need workers/agents.

Keeping step logic portable

The single most important rule: **put real work in Step subclasses, keep the

orchestrator thin.** Every option above calls the *same* Step.run(context). If

your logic lives in steps (not in Airflow operators or Prefect tasks), switching

orchestrators is glue code, not a rewrite.

Concretely:

• Steps should read inputs from context and return a plain, serializable-ish

object. They should not call print for results, reach for globals, or assume

a particular scheduler.

• Pass configuration in via constructor arguments (as the bundled steps do) so

the same class works from YAML, Airflow, or Prefect.

• At process boundaries (Airflow tasks, multi-machine runs) persist each step's

output to shared storage and pass references, not the objects themselves.

Migration checklist

When you graduate from the bare Pipeline to Airflow or Prefect:

• [ ] Confirm each step's output can be persisted (CSV/Parquet for data, joblib

or pickle for fitted models/transformers).

• [ ] Replace in-memory context passing with artifact storage + references at any

boundary that crosses a process or machine.

• [ ] Move per-step retry/timeout policy into the scheduler (retries=...) and

stop relying on fail_fast alone.

• [ ] Externalize configuration (env vars / a config store) instead of a local

configs/pipeline.yaml path.

• [ ] Add idempotency: a re-run of ingest/train should overwrite, not append.
• [ ] Keep tests/test_pipeline.py green -- the bare Pipeline remains the

fastest way to unit-test step wiring before it ever reaches a scheduler.