The Project | About Us | Contribute | Donations | License
HOME My courses Completed
◄ Previous
◄

Orchestrating ML pipelines

Next ►
►

CI and CD react to code events: a push, a PR, a tag. But module 4 left a decision waiting to be honored: the churn batch scoring must run every Monday at 06:00 to feed marketing's retention campaign, and to this day it's still "someone launches score_batch.py on Mondays". No commit triggers that — the calendar does. And scheduling it isn't enough: it must be retried if the data remote fails at 06:01, someone must be alerted if it fails for real, and on Tuesday you need to be able to answer "did the scoring run yesterday, and with what result?". That's orchestration, and this lesson solves it with Prefect: you'll understand what an orchestrator adds over cron and over dvc repro, compare the market options, and build the real flow for CineClick's weekly scoring with retries, a timeout, alerts, and a schedule.

Contents

  1. What an orchestrator is and what it adds over cron and dvc repro
  2. The orchestrator landscape and why Prefect for CineClick
  3. Prefect concepts: flow, task, deployment, schedule, work pool
  4. The real flow: flows/weekly_scoring.py
  5. Failures: retries with backoff, timeout, and alerts
  6. Scheduling Monday at 06:00
  7. A sketch of the second flow: retraining
  8. Orchestrator vs. CI/CD: what goes where

What an orchestrator is and what it adds over cron and dvc repro

The immediate temptation is a crontab: 0 6 * * 1 python scripts/score_batch.py. It works… until the first early morning when the DVC remote is slow to respond, the script dies, nobody finds out, and marketing launches Monday's campaign with last week's parquet. Cron launches processes; it doesn't operate them. The difference lies in everything that happens when something goes wrong:

Need cron dvc repro by hand Orchestrator
Scheduling (calendar) Yes No Yes
Dependencies between steps No (one monolithic script) Yes (the dvc.yaml DAG) Yes
Automatic retries with waits No No Yes, per task and configurable
Per-step timeout No No Yes
Alerts on failure Only if you rig up cron's email No Yes (Slack, webhook, email)
Historical visibility ("did it run on Monday?") Grep through syslog, with luck No UI with every run, its state and logs
Re-running only what failed No Partial (caches stages) Yes
Passing data between steps Files and faith Declared files Function return values

Mind the second column: dvc repro already gives us a reproducible DAG (module 2) and we are not throwing it away. DVC knows which steps depend on which data; what it doesn't know is when to run, how to retry when the network flickers, or how to alert when it fails. Orchestrator and DVC are complementary: the orchestrator is the alarm clock, the supervisor, and the megaphone; DVC (or our Python functions) is the work itself. In fact, the retraining sketch at the end of this lesson wraps dvc repro inside a Prefect task.

The orchestrator landscape and why Prefect for CineClick

Orchestrator Philosophy Strong at When to choose it
Airflow The veteran: declared DAGs, central scheduler, an enormous ecosystem of operators Large data teams with hundreds of DAGs and a need for integrations (BigQuery, Spark…) Airflow already exists in the company, or the volume/variety of pipelines justifies it
Prefect Pythonic: a flow is a decorated function; the minimum infrastructure is one process Small teams that want to orchestrate existing Python code without rewriting it Few pipelines, a Python team, a desire to pay for little infrastructure
Dagster Oriented around data assets: you declare what data you produce, not just what tasks run Data lineage and quality as first-class citizens; good typing The data catalog and its lineage are the heart of the problem
Kubeflow Pipelines Kubernetes-native: every step is a container Heavy ML on K8s: GPUs, distributed training, per-step isolation A corporate ML platform on Kubernetes with many teams

The choice for CineClick was already committed in module 1: Prefect. The justification, now that we know what needs orchestrating, stands on its own:

  • Minimal volume: two pipelines (weekly scoring and, later, retraining). Standing up an Airflow with its scheduler, its database, and its workers for two flows is killing flies with a cannon.
  • It's Python we already have: score_batch.py exists and works; with Prefect, turning it into a flow means adding decorators, not rewriting it in another paradigm.
  • Near-zero infrastructure: a Prefect worker is a Python process; it can run in a small pod on the cluster we already operate. With Prefect Cloud (or a lightweight self-hosted Prefect server) we get a UI, schedules, and alerts while administering almost nothing.
  • Kubeflow would be justified if we trained distributed with GPUs — and the GPU was ruled out in writing in 04-05.

Prefect concepts: flow, task, deployment, schedule, work pool

Prefect has more pieces than we need; these five cover everything we'll do:

  • Task: a Python function decorated with @task. It's the unit of retry, timeout, and logging. Rule of thumb: one task = one step you'd want to retry or see separately in the UI.
  • Flow: a function decorated with @flow that calls tasks (and optionally other flows). It defines order and dependencies — implicitly: if task B receives A's result, B waits for A.
  • Deployment: a flow registered on the Prefect server with its configuration: where it runs from, with what parameters, and on what schedule. It's what turns "a function in my repo" into "something the server knows how to launch".
  • Schedule: the deployment's calendar; we'll use a cron expression.
  • Work pool and worker: the work pool is the queue where the server drops pending work; the worker is the process (in our case, type process: a plain Python process in a pod) that listens on that queue and executes the flows. This separation means the server never executes anything itself: it only coordinates.
flowchart LR
    S[Prefect server<br/>schedules + state + UI] -->|"Monday 06:00: work available"| WP[(Work pool<br/>cineclick-pool)]
    WP -->|picks it up| W[Worker 'process'<br/>pod on the cluster]
    W -->|runs| F[flow weekly_scoring]
    F -->|states and logs| S

The real flow: flows/weekly_scoring.py

We translate the 04-01 decision (weekly batch scoring for the retention campaign) into a flow with six tasks. Each task reuses pieces that already exist: module 2's dvc pull, the pandera customer_schema from 05-01, the champion from module 3's registry, and 04-05's vectorized prediction.

# flows/weekly_scoring.py
"""Weekly churn batch scoring for the retention campaign.
Scheduled: Mondays 06:00. Output: data/churn_predictions.parquet"""
import hashlib
import subprocess
from datetime import datetime

import mlflow
import pandas as pd
import requests
from prefect import flow, task, get_run_logger

from cineclick_churn.features import build_features
from cineclick_churn.validation import customer_schema

DATA_PATH = "data/churn_customers.csv"
OUTPUT_PATH = "data/churn_predictions.parquet"
CHAMPION_URI = "models:/churn-cineclick@champion"
MARKETING_WEBHOOK = "https://hooks.slack.com/services/XXX/mock"


@task(retries=3, retry_delay_seconds=[30, 120, 300], timeout_seconds=600)
def fetch_fresh_data() -> pd.DataFrame:
    """dvc pull of the dataset. Network-fragile: 3 retries with backoff."""
    subprocess.run(["dvc", "pull", DATA_PATH], check=True)
    return pd.read_csv(DATA_PATH)


@task  # no retries: if the data is invalid, retrying won't fix it
def validate_data(df: pd.DataFrame) -> pd.DataFrame:
    """The same pandera schema as in CI, now over the full data."""
    return customer_schema.validate(df, lazy=True)


@task(retries=2, retry_delay_seconds=60, timeout_seconds=300)
def load_champion():
    """Loads models:/churn-cineclick@champion. Network-fragile (registry)."""
    return mlflow.sklearn.load_model(CHAMPION_URI)


@task
def predict_batch(model, df: pd.DataFrame) -> pd.DataFrame:
    """Vectorized prediction over the whole dataframe at once
    (lesson 04-05: no iterating over rows, ~370x faster)."""
    X = build_features(df.drop(columns=["churned"], errors="ignore"))
    probs = model.predict_proba(X)[:, 1]
    return pd.DataFrame({
        "customer_id": df["customer_id"],
        "churn_probability": probs,
        "scoring_date": datetime.now().date().isoformat(),
    })


@task
def write_predictions(predictions: pd.DataFrame) -> str:
    predictions.to_parquet(OUTPUT_PATH, index=False)
    return OUTPUT_PATH


@task(retries=3, retry_delay_seconds=30)
def notify_marketing(path: str, n_customers: int, n_at_risk: int) -> None:
    """Webhook (mock Slack): marketing knows the file is ready."""
    requests.post(MARKETING_WEBHOOK, json={
        "text": (f"Weekly churn scoring ready: {path} — "
                 f"{n_customers} customers, {n_at_risk} at risk (p >= 0.5).")
    }, timeout=10).raise_for_status()


@flow(name="weekly-churn-scoring", log_prints=True)
def weekly_scoring():
    logger = get_run_logger()

    df = fetch_fresh_data()
    df = validate_data(df)
    model = load_champion()
    predictions = predict_batch(model, df)
    path = write_predictions(predictions)

    n_at_risk = int((predictions["churn_probability"] >= 0.5).sum())
    logger.info("Scoring complete: %s rows, %s at risk", len(predictions), n_at_risk)
    notify_marketing(path, len(predictions), n_at_risk)


if __name__ == "__main__":
    weekly_scoring()   # runnable locally as is, no server needed

Points worth pausing on:

  • The flow reads like the process: fetch → validate → load champion → predict → write → notify. Dependencies aren't declared in YAML: they emerge from each task consuming the previous one's return value. If validate_data fails, nothing downstream runs.
  • The retries aren't uniform, and that's deliberate. fetch_fresh_data and load_champion fail because of the network — retrying makes sense. validate_data fails because the data is wrong — retrying the same validation on the same data wastes time and buries the alert; it must fail fast and loud.
  • Total reuse: build_features and customer_schema are exactly the same objects used by the online API, CI, and training. Batch scoring has no feature logic or data contract of its own — that was the battle won in module 3.
  • Runnable locally: python flows/weekly_scoring.py runs the whole flow with no server or worker. Debugging a Prefect flow is debugging a Python function — one of the reasons for the choice.

Failures: retries with backoff, timeout, and alerts

Anatomy of the retries

@task(retries=3, retry_delay_seconds=[30, 120, 300], timeout_seconds=600)
  • retries=3: up to 4 attempts total (the original plus 3).
  • retry_delay_seconds=[30, 120, 300]: an increasing wait (backoff) — 30 s, then 2 min, then 5 min. A transient network failure usually resolves itself; hammering every second only adds load to a service that's already suffering.
  • timeout_seconds=600: if the dvc pull hangs (worse than failing: it doesn't tell you), after 10 minutes Prefect kills the attempt and counts it as a failure — which in turn triggers the next retry.

And when it fails for real

Suppose that on Monday the storage remote is down for a full hour. The 4 attempts get exhausted (06:00, 06:10 with the timeout plus the waits…), the task ends in a Failed state, the whole flow ends Failed, and three things happen:

  1. Visible state: the run shows up in red in the Prefect UI, with each attempt's traceback and timings. On Tuesday there's nothing to reconstruct — it's written down.
  2. Alert: a Prefect automation ("when a flow run of weekly-churn-scoring enters Failed → notify the #ml-alerts channel") warns the team. The configuration is declarative on the server; nothing gets programmed into the flow. Distinguish it from notify_marketing: that task is part of the business process (announcing the parquet is ready); the automation is operations (announcing something broke).
  3. Cheap manual re-run: once the remote is fixed, one click on "Retry" in the UI (or prefect flow-run retry <id>) relaunches the run. No need to wait for next Monday or to remember the arguments.

The consequence for marketing is under control too: since the notification is only sent at the end, they never get told about a parquet that doesn't exist. No news = no new file; and with the internal alert, the team can proactively warn marketing about the delay.

Scheduling Monday at 06:00

With Prefect 2/3, the deployment is declared from the flow itself and served with a worker. First, the work pool (once):

prefect work-pool create cineclick-pool --type process

Then, register the deployment with its schedule — the cron expression 0 6 * * 1 reads: minute 0, hour 6, any day of the month, any month, day of the week 1 (Monday):

# flows/deploy_flows.py
from weekly_scoring import weekly_scoring

if __name__ == "__main__":
    weekly_scoring.deploy(
        name="weekly-scoring-monday",
        work_pool_name="cineclick-pool",
        cron="0 6 * * 1",              # Mondays 06:00
        # the schedule's timezone is set to Europe/Madrid on the
        # server; without it, "06:00" would be UTC and in summer would land at 08:00
    )

And a worker listening on the pool (in production, a small pod on the cluster with the repo installed; in development, your terminal):

prefect worker start --pool cineclick-pool

With this, the decision made in 04-01 — "weekly scoring, Mondays at 06:00" — finally honors itself: the server creates the run at the scheduled time, the worker picks it up, and if nobody is watching… that's precisely what the states and alerts are for.

A sketch of the second flow: retraining

Scoring isn't the only recurring process on the horizon: sooner or later the model will have to be retrained on fresh data. The flow is structurally similar and reuses even more — it wraps module 2's entire DVC pipeline:

# flows/retraining.py — SKETCH, completed in module 6
from prefect import flow, task
import subprocess


@task(retries=2, retry_delay_seconds=60)
def fresh_data():
    subprocess.run(["dvc", "pull"], check=True)


@task(timeout_seconds=3600)
def reproduce_pipeline():
    """dvc repro runs prepare_data -> build_features ->
    train -> evaluate, with its stage cache intact."""
    subprocess.run(["dvc", "repro"], check=True)


@task
def register_challenger():
    """Reads metrics.json; if it clears the 05-01 thresholds, registers the
    new version in MLflow and gives it the @challenger alias.
    NEVER @champion: promotion follows the human flow from 05-02."""
    ...


@flow(name="churn-retraining")
def retraining():
    fresh_data()
    reproduce_pipeline()
    register_challenger()

Notice what this sketch decides and what it does not decide. It decides the how: DVC does the heavy lifting (with its stage cache: if the data didn't change, prepare_data doesn't even run), and the result goes at most to @challenger — promotion to @champion keeps 05-02's human approval, even when training is automatic. What it doesn't decide is the when: every month? When data drift crosses a threshold? When performance drops? That policy is the subject of lesson 06-03, once we have the monitoring that informs it. For now, the flow exists and can be launched by hand from the UI — which is already infinitely better than "Laura launches it from her laptop".

Orchestrator vs. CI/CD: what goes where

With three automation tools in play (CI, CD, orchestrator) it's easy to put things where they don't belong. The dividing criterion is the trigger:

Process Trigger Tool Why
Code/data/model tests Push, PR GitHub Actions (CI) Code event; ephemeral; per commit
Building and deploying the image Tag v*.*.* GitHub Actions (CD) Code event; software delivery
Weekly batch scoring Calendar (Monday 06:00) Prefect Recurring data process; needs retries, state, and alerts
Retraining Calendar or a monitoring signal (06-03) Prefect Long data process, with dependencies and retries
Promoting a model to @champion Human decision Script + review (05-02) Neither a code event nor a calendar: it's a decision

The mnemonic: GitHub Actions answers "the code changed"; Prefect answers "it's time to process data". Actions runners are ephemeral and anonymous — perfect for tests, terrible for a business process whose run history you'll want to consult in three months. You could force a schedule: in Actions for the scoring (it exists), but you'd lose per-step retries, per-step timeouts, selective re-runs, and the state UI — that is, everything we asked for at the start of the lesson.

Common Mistakes and Tips

  • A flow with one giant task. If weekly_scoring were a single task doing everything, a webhook failure at the end would force repeating the download and the prediction too. Task granularity defines retry granularity: separate the fragile (network) from the deterministic (computation).
  • Retries on everything "just in case". Retrying a failed data validation delays the alert by 10 minutes only to hit the same error. Retry the transient (network, external services); fail fast on the deterministic.
  • Forgetting the schedule's timezone. 0 6 * * 1 in UTC is 07:00 or 08:00 in Madrid depending on the time of year — with marketing waiting since 06:00. Explicitly pin Europe/Madrid on the deployment's schedule.
  • Business logic inside the flow instead of the package. If predict_batch reimplemented the features "because it was more convenient", we'd have a third copy of the logic module 3 unified. Tasks orchestrate; the cineclick_churn package implements.
  • Confusing the business notification with the failure alert. notify_marketing announces success to a consumer; the Prefect automation announces failure to the team. If you mix both into the same task, a scoring failure can end up telling marketing "file ready", or leave the team without an alert.
  • Not testing the flow locally before deploying it. The flow is a function: python flows/weekly_scoring.py runs it whole. Debugging by staring at worker logs on the server is the slow way.
  • Tip: give the flow a stable, descriptive name (weekly-churn-scoring) from day one. Automations, dashboards, and incident conversations hang off that name; renaming it later breaks more than it seems.

Exercises

Exercise 1

Marketing asks that, besides the parquet, the scoring also write a CSV with only the 500 customers with the highest churn probability (data/top_risk.csv), and that if this extra step fails, the flow still counts as successful (the parquet is what's critical; the CSV is a courtesy). Write the task and its integration into the flow. Hint: look into return_state=True or exception handling inside the flow.

Exercise 2

On Monday at 06:00 the MLflow server is down for unannounced maintenance and comes back at 06:20. With the flow written as it is, does Monday's scoring complete or fail? Justify it by working out the timeline of load_champion's attempts (retries=2, retry_delay_seconds=60, timeout_seconds=300), assuming each attempt against the down server fails after ~10 seconds and that the previous tasks finished at 06:05.

Exercise 3

A teammate proposes moving CI into the orchestrator: "that way everything lives in Prefect, a ci flow that runs ruff and pytest whenever someone pushes". Give three concrete reasons why the tests from lesson 05-01 fit GitHub Actions better than Prefect.

Solutions

Solution 1

@task(retries=1, retry_delay_seconds=30)
def write_top_risk(predictions: pd.DataFrame, n: int = 500) -> str:
    top = predictions.nlargest(n, "churn_probability")
    top.to_csv("data/top_risk.csv", index=False)
    return "data/top_risk.csv"

And in the flow, after write_predictions:

    state = write_top_risk(predictions, return_state=True)
    if state.is_failed():
        logger.warning("top_risk.csv not generated; the parquet IS available")

With return_state=True the call returns the state instead of propagating the exception: the task shows up red in the UI (visibility intact) but the flow continues and finishes successfully. It's the idiomatic way to mark a step as "desirable but not critical". A surrounding try/except would work too, but the flow's state would be less faithful in the UI.

Solution 2

It fails. Timeline: previous tasks done at 06:05. Attempt 1 of load_champion ≈ 06:05:00, fails after ~10 s (06:05:10). Wait 60 s → attempt 2 ≈ 06:06:10, fails (06:06:20). Wait 60 s → attempt 3 (the last: retries=2 = 3 attempts total) ≈ 06:07:20, fails (06:07:30). Result: all three attempts are spent by 06:07:30 and the server doesn't come back until 06:20 — the flow fails, the automation alerts, and at 06:20+ someone (or one click on Retry) relaunches it. A double moral: (a) work out the total window your retries cover (here ~2.5 minutes, insufficient for a 20-minute maintenance); for dependencies with known long outages, a longer backoff ([60, 300, 900]) covers ~21 minutes; (b) even when the retries aren't enough, the system degraded well: visible state, alert emitted, one-click re-run — compare that with the silent cron from the beginning.

Solution 3

(1) Trigger: CI reacts to push/PR, an event GitHub emits natively to Actions; moving it to Prefect would require rigging webhooks and would lose the integration with the review flow. (2) Merge blocking: CI's value is marking the PR green/red and blocking the merge (branch protection); Prefect plays no part in that mechanism — you'd have tests that run but protect nothing. (3) Ephemeral environment per commit: every Actions run starts from a clean runner and checks out the PR's exact commit, which is exactly what you want for validating code; the Prefect worker is a persistent process with one version of the repo installed, designed for data processes, not for testing N concurrent branches. It's the lesson's table in action: code event → Actions; recurring data process → Prefect.

Conclusion

CineClick's last manual process now moves on its own: the weekly-churn-scoring flow (in flows/weekly_scoring.py) pulls the data with dvc pull, validates it with the same pandera schema as CI, loads models:/churn-cineclick@champion, predicts in a vectorized batch, writes data/churn_predictions.parquet, and notifies marketing — every Monday at 06:00 (0 6 * * 1, Europe/Madrid) via a deployment on the cineclick-pool work pool, with backoff retries where there's network, timeouts where things can hang, and an alert to the team when it fails for real. We chose Prefect over Airflow, Dagster, and Kubeflow because of what CineClick is: two pipelines, a small team, and Python code that already worked. And the automation map got divided up: Actions for code events, Prefect for data processes, and the decisions — promotion to @champion, even from the sketched retraining flow — for people. The module now has CI, CD, and orchestration; its most delicate piece remains. Because everything we've built deploys the new model all at once after staging, and offline metrics — however good — don't guarantee what it will do with real traffic. The module's final lesson teaches how to release a model with a safety net: shadow, canary, and A/B.

◄ Previous
◄
Next ►
►
The Project | About Us | Contribute | Donations | License
© Copyright 2026. All rights reserved