Prefect

À quoi ça sert

Tu écris un pipeline d'entraînement : charger les données → nettoyer → entraîner → évaluer → enregistrer le modèle. Tant qu'il tourne dans un notebook, ça va. Le jour où tu veux le lancer toutes les nuits, qu'il réessaie tout seul si l'API source plante, qu'il t'alerte en cas d'échec et que tu puisses voir l'historique des runs, il te faut un orchestrateur.

Prefect te donne :

• Tasks et Flows — décore tes fonctions, le code reste du Python normal. Pas de DSL exotique.
• Planification — cron, intervalle, événement.
• Retries automatiques avec backoff configurable.
• Caching — réutilise le résultat d'une tâche si les entrées n'ont pas changé.
• UI — DAG visuel, logs par tâche, durée, statut, historique des runs.
• Deployments — tu déclares « ce flow tourne sur cette infra avec cette planif », et un worker Prefect le lance le moment venu.

Un exemple d'usage

Pipeline d'entraînement minimal : charger l'iris, splitter, entraîner, évaluer, logger dans MLflow.

from prefect import flow, task
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import accuracy_score
import mlflow

@task(retries=2, retry_delay_seconds=5)
def load_data():
    return load_iris(return_X_y=True)

@task
def split(X, y):
    return train_test_split(X, y, random_state=42)

@task
def train(X_train, y_train, depth: int):
    m = DecisionTreeClassifier(max_depth=depth, random_state=42)
    m.fit(X_train, y_train)
    return m

@task
def evaluate(model, X_test, y_test) -> float:
    return accuracy_score(y_test, model.predict(X_test))

@flow(name="iris-training")
def training_pipeline(depth: int = 5):
    X, y = load_data()
    X_train, X_test, y_train, y_test = split(X, y)
    model = train(X_train, y_train, depth)
    acc = evaluate(model, X_test, y_test)

    with mlflow.start_run():
        mlflow.log_param("max_depth", depth)
        mlflow.log_metric("accuracy", acc)
        mlflow.sklearn.log_model(model, name="model")

    return acc

if __name__ == "__main__":
    training_pipeline(depth=5)

Lance avec uv run python pipeline.py : ça tourne comme un script Python normal, avec en bonus des logs structurés, des retries sur load_data, et un flow run visible dans l'UI Prefect.

How-to : démarrer avec Prefect

1. Installer Prefect

   bash Copier

   uv add prefect

   Vérifie l'installation :

   bash Copier

   uv run prefect version

   Cette fiche couvre Prefect 3.x. Si tu tombes sur des tutos qui parlent de DaskTaskRunner ou de storage, c'est probablement de la 1.x — vérifie la doc avant de copier.

2. Lancer le serveur Prefect en local

   Sans serveur, tes flows tournent quand même mais sans UI ni historique. Pour avoir tout ça en local :

   bash Copier

   uv run prefect server start

   Ça démarre un serveur sur http://127.0.0.1:4200. Laisse-le tourner et ouvre l'URL : tu vois une UI avec Dashboard, Flows, Deployments.

   Self-hosted vs Prefect Cloud

   Prefect Cloud (SaaS) propose un free tier généreux et évite d'héberger toi-même. Pour un projet de cours, le serveur local suffit. Tu bascules entre les deux avec prefect cloud login et prefect profile use ....

3. Connecter ton code au serveur

   Dans un autre terminal, dis à Prefect où trouver l'API :

   bash Copier

   uv run prefect config set PREFECT_API_URL=http://127.0.0.1:4200/api

   Lance ton pipeline.py : un nouveau flow run apparaît dans l'UI, avec le DAG des tâches, leurs durées et leurs logs.

4. Comprendre tasks vs flows

   • @task — une unité de travail. Bénéficie des retries, du caching, de la traçabilité individuelle.
   • @flow — une fonction qui orchestre des tasks (et éventuellement des sous-flows). C'est l'unité que tu planifies et que tu exécutes.

   Règle simple : appels longs ou faillibles (I/O, DB, API externe, calcul) → tâches. Le flow reste mince et lit l'orchestration.

5. Retries et caching

   python Copier

   from datetime import timedelta
   from prefect import task
   from prefect.cache_policies import INPUTS
   
   @task(retries=3, retry_delay_seconds=[1, 5, 15])
   def fetch_api(url: str):
       return requests.get(url).json()
   
   @task(cache_policy=INPUTS, cache_expiration=timedelta(hours=1))
   def load_dataset(name: str):
       return expensive_load(name)

   Backoff exponentiel + caching par valeur d'entrée : si tu relances le flow avec le même name dans l'heure, load_dataset ne re-télécharge pas, il rejoue son résultat caché.

6. Planifier un flow (Deployment)

   Pour un flow qui doit tourner toutes les nuits sans que tu sois là, on crée un deployment :

   python (en bas de pipeline.py) Copier

   if __name__ == "__main__":
       training_pipeline.serve(
           name="nightly-training",
           cron="0 3 * * *",         # 3h du matin
           parameters={"depth": 5},
       )

   serve() garde le process ouvert et joue le rôle de worker : à 3h, il déclenche le flow. Tu vois le deployment dans l'UI, tu peux le déclencher à la main avec le bouton « Run » et passer d'autres paramètres.

7. Notifications en cas d'échec

   Dans l'UI, Automations → New Automation. Tu choisis un trigger (ex. Flow run enters Failed state) et une action (e-mail, webhook Discord/Slack). Pas de code à écrire. C'est ce qui remplace les alertes maison à coup de try/except.

8. En production sur Docker / Kubernetes

   Pour exécuter un flow dans un conteneur frais à chaque run (idéal en prod), on déclare un work pool de type docker ou kubernetes, on push une image avec ton code, et un worker Prefect tire les runs et lance les conteneurs. La logique reste la même côté code, seul le déploiement change.

Aide-mémoire

from prefect import flow, task, get_run_logger

@task(retries=2, retry_delay_seconds=5)
def step(x): return x * 2

@flow(name="demo", log_prints=True)
def demo(n: int = 3):
    log = get_run_logger()
    log.info(f"start n={n}")
    return [step(i) for i in range(n)]

# Planifier
demo.serve(name="demo-deploy", cron="*/5 * * * *")

prefect server start
prefect config set PREFECT_API_URL=http://127.0.0.1:4200/api
prefect deployment ls
prefect deployment run "iris-training/nightly-training"
prefect flow-run ls --limit 10
prefect work-pool ls
prefect worker start --pool default

Prefect et le reste de l'écosystème

• MLflow — combo naturel. Prefect orchestre l'entraînement, MLflow logge le run. Pattern type pour le projet de fin de formation : un flow Prefect planifié qui charge les données, entraîne, évalue, logge dans MLflow, et promeut le modèle si la métrique dépasse un seuil.
• Celery — voir l'encart en haut. Pas concurrents : Celery décharge des routes HTTP, Prefect orchestre des pipelines.
• Docker / Kubernetes — les work pools Docker/K8s exécutent chaque run dans un conteneur frais. Idéal pour des dépendances lourdes ou des modèles volumineux.
• FastAPI — depuis une route, tu peux déclencher un deployment Prefect via son API REST. Mais si la fonction est rapide, Celery est plus adapté.
• pandas / SQLAlchemy — Prefect brille sur les pipelines ETL : extract via SQLAlchemy, transform en pandas, load en base, le tout planifié et observable.
• Prometheus + Grafana — l'UI Prefect suffit pour suivre les flows. Si tu veux les chiffres dans Grafana, le serveur Prefect expose une API REST que tu peux scraper, ou tu loggues toi-même les métriques métier dans Prometheus depuis tes tasks.

Pour aller plus loin

• Documentation officielle : docs.prefect.io
• Quickstart : docs.prefect.io/v3/get-started/quickstart
• Repo GitHub : github.com/PrefectHQ/prefect
• Recipes : github.com/PrefectHQ/prefect-recipes