Pipeline orchestration with Prefect

Look at everything this series has built so far, laid end to end. Pull the data with dvc pull. Run train_experiments.py, which trains several models and logs every one of them to MLflow. Find the best run and register it. Promote it to @champion. Rebuild the Docker image. Redeploy the API.

Every single piece of that chain is solid, tested, and versioned. None of it runs by itself. A person has to remember the order, remember to run it again when new data shows up, and notice if a step fails partway through. That's not a tooling gap the same way the earlier ones were, Git, DVC, Docker, and MLflow each solved a well-defined problem, but nothing so far has solved "and now run all of this, in order, on its own, and tell me if something breaks."

That's orchestration. This article covers Prefect, a Python-native tool for turning exactly this kind of manual chain into something that runs unattended, retries when something flaky happens, and keeps a record of every attempt.

---

What this article covers

#	Topic
1	The problem: a correct pipeline nobody's running
2	What Prefect is, and why not just write a bash script
3	Core concepts: tasks, flows, and automatic dependencies
4	Installing Prefect and running your first flow
5	Retries: making failure someone else's problem
6	Caching: skipping work that's already been done
7	The practical part: orchestrating the churn training pipeline
8	Seeing it in the dashboard
9	Scheduling: from manual command to weekly retrain
10	Where this leaves the whole system

---

Part 1 — The problem: a correct pipeline nobody's running

Nothing in this series has been wrong, exactly. Article 8 built a training script that logs everything properly. Article 9 built a promotion step that's clean and auditable. The gap is what sits between them and reality: someone has to actually run this sequence, notice if load_data fails because a file moved, remember that promotion should only happen after training succeeds, and do all of this again next week when fresh customer data shows up.

That's a real cost, and it's an easy one to underestimate, because it doesn't show up as a bug in any single piece. It shows up as: nobody retrained the model in six weeks, or someone ran the promotion step before checking the new run's accuracy, or a transient network hiccup during data loading silently killed the whole pipeline and nobody noticed until a stakeholder asked why predictions looked stale.

---

Part 2 — What Prefect is, and why not just write a bash script

Prefect is a Python-native workflow orchestration tool: you write ordinary Python functions, add two decorators, and Prefect handles scheduling, retries, dependency tracking, and logging around them.¹ It was founded in 2018 by Jeremiah Lowin, and the core engine has been Apache 2.0 open source from the start.

The honest question worth asking first: why not just write a bash script that runs the steps in order? For a one-off, that's fine. It stops being fine the moment you need any of the following, all of which a bash script gives you nothing for: automatic retries when a step fails for a transient reason, a record of every past run and whether it succeeded, alerting when something breaks, and running the same pipeline on a schedule without a cron job silently failing into a black hole.

Prefect's specific pitch, compared to older orchestration tools like Airflow, is that you don't learn a new way to define pipelines. You write normal Python functions and call them normally. Two decorators, @task and @flow, are the entire vocabulary.

Airflow remains a completely reasonable choice, especially at large organizations already running it. This article uses Prefect because "write ordinary Python" fits a series built on the principle that you shouldn't need to learn a new language to solve each new problem.

---

Part 3 — Core concepts: tasks, flows, and automatic dependencies

Two decorators carry almost the whole idea.

A task is one unit of work, a Python function decorated with @task. Loading data, training a model, promoting a model, each of these is naturally one task.

A flow is the pipeline itself, a Python function decorated with @flow that calls your tasks. Prefect watches how data moves between task calls inside a flow and builds the dependency graph automatically. Call transform(extract()), and Prefect already knows extract has to finish before transform starts. No separate dependency declaration, no >> operator, nothing to keep in sync with the actual code.

Each task run and flow run gets its own record: what it was called, when it started and finished, what state it ended in, and its full log output. That record is the thing a bash script never gives you.

---

Part 4 — Installing Prefect and running your first flow

pip install prefect

from prefect import task, flow

@task
def say_hello(name):
    print(f"Hello, {name}!")
    return f"Hello, {name}!"

@flow
def greet_flow():
    return say_hello("MLOps")

if __name__ == "__main__":
    greet_flow()

python greet_flow.py

06:27:43 | INFO | prefect - Starting temporary server on http://127.0.0.1:8111
06:27:55 | INFO | Flow run 'spirited-bullfinch' - Beginning flow run 'spirited-bullfinch' for flow 'greet-flow'
06:27:55 | INFO | Task run 'say_hello-0da' - Finished in state Completed()
06:27:56 | INFO | Flow run 'spirited-bullfinch' - Finished in state Completed()
06:27:56 | INFO | prefect - Stopping temporary server on http://127.0.0.1:8111

Confirmed by actually running it: a plain python call with no setup, no separate server process started beforehand, and it works. Prefect spins up a lightweight temporary server automatically to record the run, then shuts it down when the script finishes. Every flow run also gets a randomly generated two-word name, spirited-bullfinch here, so you can talk about a specific run without needing to remember a UUID.

For a persistent dashboard you can revisit later, run prefect server start once, which brings up a UI at http://127.0.0.1:4200 showing every flow and task run, its state, and its logs.

---

Part 5 — Retries: making failure someone else's problem

This is the single feature that most directly answers "what does orchestration actually buy me."

from prefect import task, flow

@task(retries=3, retry_delay_seconds=1)
def flaky_task():
    # simulates a transient failure, like a dropped network connection
    ...

@flow
def retry_flow():
    return flaky_task()

Tested directly against a task engineered to fail twice before succeeding:

Task run 'flaky_task-0e8' - Task run failed with exception: ValueError('Simulated transient failure') - Retry 1/3 will start 1 second(s) from now
Task run 'flaky_task-0e8' - Task run failed with exception: ValueError('Simulated transient failure') - Retry 2/3 will start 1 second(s) from now
Task run 'flaky_task-0e8' - Finished in state Completed()

retries=3, retry_delay_seconds=1 is the entire configuration. No try/except block, no manual retry loop, no custom logging to explain what happened. For an ML pipeline specifically, this matters more than it might look: a data source that's briefly unavailable, an MLflow server that hiccups for a second, a flaky network call, none of these should mean the whole night's training run silently never happened.

That's the recovery case. It's worth seeing the other side too, what happens when a task is genuinely broken, not just flaky, and exhausts every retry:

@task(retries=2, retry_delay_seconds=1)
def always_fails():
    raise ValueError("This step is genuinely broken, not just flaky")

@flow
def failure_flow():
    result = always_fails()
    return downstream_step(result)  # should never run

Confirmed by actually running it: after both retries fail, the flow ends in a Failed state, and, this is the part that matters, downstream_step never runs at all. Its log line never appears, not even once. For the churn pipeline, that's the exact safety property worth having: if train_and_log genuinely fails, promote_best_model should never execute. Prefect enforces that by default, with no extra code written for it.

---

Part 6 — Caching: skipping work that's already been done

Article 3 taught this exact idea already, just for a different tool. dvc repro skips a pipeline stage if nothing it depends on has changed. Prefect gives individual tasks the same behavior.

from prefect.tasks import task_input_hash
from datetime import timedelta

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(minutes=10))
def expensive_step(x):
    print("actually running")
    return x * 2

cache_key_fn=task_input_hash tells Prefect to fingerprint the task's inputs. Call it again with the same inputs inside the cache window, and Prefect returns the stored result instead of re-running the function. Confirmed directly: calling a cached task twice with identical input runs the underlying function exactly once, total, and the second call's log shows the task finishing in a Cached state rather than Completed.

For the churn flow, this is most useful on load_data. If the raw CSV hasn't changed, there's no reason to re-read and re-clean 7,043 rows every time the flow runs, especially on a schedule where most weeks nothing about the source file has actually moved.

---

Part 7 — The practical part: orchestrating the churn training pipeline

This is Article 8's track_experiments.py and Article 9's promotion step, rebuilt as one Prefect flow. Every number below is real, produced by actually running this exact code against the same churn dataset used throughout the series.

"""orchestrate_training.py — Prefect flow that trains, tracks, and promotes the churn model."""
import pandas as pd
import mlflow
import mlflow.sklearn
from mlflow import MlflowClient
from prefect import task, flow, get_run_logger
from sklearn.model_selection import train_test_split
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score


@task(retries=2, retry_delay_seconds=5)
def load_data(path="data/telco_churn.csv"):
    logger = get_run_logger()
    df = pd.read_csv(path)
    df["TotalCharges"] = pd.to_numeric(df["TotalCharges"], errors="coerce").fillna(0)
    df = df.drop(columns=["customerID"])
    df["Churn"] = df["Churn"].map({"Yes": 1, "No": 0})
    logger.info(f"Loaded {len(df)} rows")
    return df


@task
def split_data(df):
    X = df.drop(columns=["Churn"])
    y = df["Churn"]
    numeric_features = ["SeniorCitizen", "tenure", "MonthlyCharges", "TotalCharges"]
    categorical_features = [c for c in X.columns if c not in numeric_features]
    preprocessor = ColumnTransformer(transformers=[
        ("num", StandardScaler(), numeric_features),
        ("cat", OneHotEncoder(handle_unknown="ignore"), categorical_features),
    ])
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )
    return preprocessor, X_train, X_test, y_train, y_test


@task
def train_and_log(run_name, clf, params, preprocessor, X_train, X_test, y_train, y_test):
    logger = get_run_logger()
    with mlflow.start_run(run_name=run_name):
        mlflow.log_params(params)
        pipe = Pipeline([("preprocessor", preprocessor), ("classifier", clf)])
        pipe.fit(X_train, y_train)
        pred = pipe.predict(X_test)

        acc = accuracy_score(y_test, pred)
        mlflow.log_metric("accuracy", acc)
        mlflow.log_metric("precision", precision_score(y_test, pred))
        mlflow.log_metric("recall", recall_score(y_test, pred))
        mlflow.log_metric("f1", f1_score(y_test, pred))
        mlflow.sklearn.log_model(pipe, name="model")

        logger.info(f"{run_name}: accuracy={acc:.4f}")
        return acc


@task
def promote_best_model(experiment_name="churn-prediction", model_name="ChurnModel"):
    logger = get_run_logger()
    client = MlflowClient()
    experiment = client.get_experiment_by_name(experiment_name)
    runs = client.search_runs(
        experiment_ids=[experiment.experiment_id], order_by=["metrics.accuracy DESC"]
    )
    best_run = runs[0]
    logger.info(f"Best run: {best_run.data.tags.get('mlflow.runName')} "
                f"(accuracy={best_run.data.metrics['accuracy']:.4f})")

    model_uri = f"runs:/{best_run.info.run_id}/model"
    registered = mlflow.register_model(model_uri=model_uri, name=model_name)
    client.set_registered_model_alias(model_name, "champion", registered.version)
    logger.info(f"Promoted {model_name} version {registered.version} to @champion")
    return registered.version


@flow(name="churn-training-pipeline")
def churn_training_flow():
    mlflow.set_experiment("churn-prediction")

    df = load_data()
    preprocessor, X_train, X_test, y_train, y_test = split_data(df)

    configs = [
        ("logreg_default", LogisticRegression(max_iter=1000, random_state=42),
         {"model_type": "LogisticRegression", "max_iter": 1000}),
        ("rf_200_depth10", RandomForestClassifier(n_estimators=200, max_depth=10, random_state=42),
         {"model_type": "RandomForest", "n_estimators": 200, "max_depth": 10}),
        ("gb_200_lr005", GradientBoostingClassifier(n_estimators=200, learning_rate=0.05, random_state=42),
         {"model_type": "GradientBoosting", "n_estimators": 200, "learning_rate": 0.05}),
    ]

    for run_name, clf, params in configs:
        train_and_log(run_name, clf, params, preprocessor, X_train, X_test, y_train, y_test)

    promote_best_model()


if __name__ == "__main__":
    churn_training_flow()

python orchestrate_training.py

Task run 'load_data-131' - Loaded 7043 rows
Task run 'split_data-f41' - Finished in state Completed()
Task run 'train_and_log-3c5' - logreg_default: accuracy=0.8055
Task run 'train_and_log-20f' - rf_200_depth10: accuracy=0.7999
Task run 'train_and_log-d02' - gb_200_lr005: accuracy=0.8048
Task run 'promote_best_model-ea2' - Best run: logreg_default (accuracy=0.8055)
Task run 'promote_best_model-ea2' - Promoted ChurnModel version 1 to @champion
Flow run 'chivalrous-swallow' - Finished in state Completed()

Every number matches the series exactly, logreg_default at 0.8055, rf_200_depth10 at 0.7999, gb_200_lr005 at 0.8048, because it's the same pipeline from Article 8 producing the same results, just no longer requiring a person to babysit it. promote_best_model correctly picked logreg_default, exactly as Article 9's manual version did, and promoted it to @champion the same way.

A few details worth pointing out. @task(retries=2, retry_delay_seconds=5) on load_data specifically, not on every task, is a deliberate choice: data loading is the step most likely to fail for a transient reason (a network drive, a slow disk), so it's the one worth retrying automatically. get_run_logger() gives structured, per-task logs for free, logger.info(...) inside a task shows up tagged with that exact task run's name in the output above, no manual log formatting needed. And this flow only shows three of the six configurations from Article 8 for readability, extending the configs list to all six, or replacing the loop with train_and_log.map(configs, ...) to fan them out, follows exactly the same pattern.

---

Part 8 — Seeing it in the dashboard

Everything up to this point ran with Prefect's temporary, disposable server, spun up per script and torn down when it finishes. Run prefect server start once instead, and every flow run gets recorded permanently, queryable at http://127.0.0.1:4200, or through the same REST API the dashboard itself uses underneath.

Querying it directly after running the churn flow a couple of times, alongside one of the failures from Part 5:

name='frisky-beluga'   state=COMPLETED   start=2026-08-13T07:51:11
name='meek-chipmunk'   state=COMPLETED   start=2026-08-13T07:50:01
name='marble-cockle'   state=FAILED      start=2026-08-13T07:49:21

A failed run and two successful ones, sitting in the same permanent list, each one inspectable down to the individual task level:

task=load_data              state=COMPLETED
task=split_data             state=COMPLETED
task=train_and_log          state=COMPLETED
task=train_and_log          state=COMPLETED
task=train_and_log          state=COMPLETED
task=promote_best_model     state=COMPLETED

That's the entire pitch from Part 1, made concrete. A bash script gives you none of this. The failed run from three minutes ago doesn't vanish the moment the terminal scrolls past it. It's sitting right there, permanently, next to the runs that worked, exactly the record that used to only exist if someone happened to be watching when it broke.

---

Part 9 — Scheduling: from manual command to weekly retrain

Everything above still starts with a person typing python orchestrate_training.py. The last piece is removing that.

if __name__ == "__main__":
    churn_training_flow.serve(
        name="weekly-churn-retrain",
        cron="0 2 * * 1",  # every Monday at 2am
    )

.serve(...) keeps a lightweight process running that watches the schedule and kicks off churn_training_flow automatically, on time, logging every run the same way the manual runs above were logged. cron="0 2 * * 1" is a standard cron expression, every Monday at 2am, matching the retraining cadence Article 0 originally described as a hypothetical example back at the start of this series. Swap the cron string for interval=timedelta(days=7) if a plain "every N days" schedule reads more clearly for your team than cron syntax does.

For running scheduled flows reliably in production, beyond a single always-on process, Prefect's production model uses work pools: a work pool is deployed once, and specifies where flow runs should actually execute, a Docker container, a Kubernetes cluster, cloud infrastructure, and workers polling that pool pick up and run scheduled work as it comes due.² That's a natural fit for the next article in this series: a work pool that hands each scheduled run off to the same Docker image built in Article 5.

---

Part 10 — Where this leaves the whole system

Every tool in this series now has exactly one job, and Prefect's job is making sure the other jobs actually happen without a person remembering to trigger them. Git versions code. DVC versions data and models. Docker guarantees the environment. FastAPI validates and serves. MLflow Tracking records every experiment. The Model Registry manages which model is live. Prefect runs the whole training-through-promotion sequence on a schedule, retrying transient failures automatically, skipping work that's already been done, and stopping cleanly, before promotion, when something is genuinely broken rather than just flaky. Every attempt, successful or not, sits permanently in a queryable history, the same way DVC keeps every data version and MLflow keeps every experiment.

The chain from Part 1, pull data, train, track, register, promote, is no longer something a person has to remember. It's a flow that runs itself every Monday at 2am and tells you, honestly, whether it worked.

What's still missing is the other half of automation: nothing yet rebuilds the Docker image or redeploys the API when a new model gets promoted. That's CI/CD, and it's the article this series goes to next, GitHub Actions, triggering exactly that rebuild-and-redeploy step the moment @champion changes.