El Proyecto | Sobre nosotros | Contribuir | Donaciones | Licencia
HOME Mis cursos Finalizados
◄ Anterior
◄

Orquestación de pipelines de ML

Siguiente ►
►

CI y CD reaccionan a eventos de código: un push, una PR, un tag. Pero en el módulo 4 quedó una decisión pendiente de cumplir: el scoring batch de churn debe ejecutarse cada lunes a las 06:00 para alimentar la campaña de retención de marketing, y a día de hoy sigue siendo "alguien lanza score_batch.py los lunes". Ningún commit dispara eso — lo dispara el calendario. Y no basta con programarlo: hay que reintentarlo si el remote de datos falla a las 06:01, avisar a alguien si falla de verdad, y poder responder el martes a la pregunta "¿corrió el scoring ayer y con qué resultado?". Eso es orquestación, y esta lección la resuelve con Prefect: entenderás qué añade un orquestador sobre cron y sobre dvc repro, compararás las opciones del mercado, y construirás el flow real del scoring semanal de CineClick con reintentos, timeout, alertas y schedule.

Contenido

  1. Qué es un orquestador y qué añade sobre cron y dvc repro
  2. Panorama de orquestadores y por qué Prefect para CineClick
  3. Conceptos de Prefect: flow, task, deployment, schedule, work pool
  4. El flow real: flows/scoring_semanal.py
  5. Fallos: retries con backoff, timeout y alertas
  6. Programar el lunes a las 06:00
  7. Esbozo del segundo flow: reentrenamiento
  8. Orquestador vs. CI/CD: qué va en cada sitio

Qué es un orquestador y qué añade sobre cron y dvc repro

La tentación inmediata es un crontab: 0 6 * * 1 python scripts/score_batch.py. Funciona… hasta la primera madrugada en que el remote de DVC tarda en responder, el script muere, nadie se entera, y marketing lanza la campaña del lunes con el parquet de la semana pasada. Cron lanza procesos; no los opera. La diferencia está en todo lo que pasa cuando algo no va bien:

Necesidad cron dvc repro a mano Orquestador
Planificación (calendario) Sí No Sí
Dependencias entre pasos No (un script monolítico) Sí (el DAG de dvc.yaml) Sí
Reintentos automáticos con espera No No Sí, por task y configurable
Timeout por paso No No Sí
Alertas al fallar Solo si te montas el email de cron No Sí (Slack, webhook, email)
Visibilidad histórica ("¿corrió el lunes?") Grep en syslog, con suerte No UI con cada ejecución, estado y logs
Reejecutar solo lo que falló No Parcial (cachea stages) Sí
Paso de datos entre pasos Ficheros y fe Ficheros declarados Valores de retorno de funciones

Ojo a la segunda columna: dvc repro ya nos da un DAG reproducible (módulo 2) y no lo vamos a tirar. DVC sabe qué pasos dependen de qué datos; lo que no sabe es cuándo ejecutarse, reintentar si la red parpadea ni avisar si falla. Orquestador y DVC son complementarios: el orquestador es el despertador, el supervisor y el megáfono; DVC (o nuestras funciones Python) es el trabajo en sí. De hecho, el esbozo de reentrenamiento del final de esta lección envuelve dvc repro dentro de una task de Prefect.

Panorama de orquestadores y por qué Prefect para CineClick

Orquestador Filosofía Fuerte en Cuándo elegirlo
Airflow El veterano: DAGs declarados, scheduler central, ecosistema enorme de operadores Equipos de datos grandes con cientos de DAGs y necesidad de integraciones (BigQuery, Spark…) Ya hay Airflow en la empresa, o el volumen/variedad de pipelines lo justifica
Prefect Pythónico: un flow es una función decorada; la infraestructura mínima es un proceso Equipos pequeños que quieren orquestar código Python existente sin reescribirlo Pocos pipelines, equipo Python, ganas de pagar poca infraestructura
Dagster Orientado a assets de datos: declaras qué datos produces, no solo qué tareas corren Linaje y calidad de datos como ciudadanos de primera; buen tipado El catálogo de datos y su linaje son el centro del problema
Kubeflow Pipelines Nativo de Kubernetes: cada paso es un contenedor ML pesado sobre K8s: GPUs, entrenamiento distribuido, aislamiento por paso Plataforma ML corporativa sobre Kubernetes con muchos equipos

La elección para CineClick ya quedó comprometida en el módulo 1: Prefect. La justificación, ahora que sabemos lo que hay que orquestar, se sostiene sola:

  • Volumen mínimo: dos pipelines (scoring semanal y, más adelante, reentrenamiento). Levantar un Airflow con su scheduler, su base de datos y sus workers para dos flows es matar moscas a cañonazos.
  • Es Python del que ya tenemos: score_batch.py existe y funciona; con Prefect, convertirlo en flow es añadir decoradores, no reescribirlo en otro paradigma.
  • Infraestructura casi nula: un worker de Prefect es un proceso Python; puede correr en un pod pequeño del clúster que ya operamos. Con Prefect Cloud (o un servidor Prefect self-hosted ligero) tenemos UI, schedules y alertas sin administrar apenas nada.
  • Kubeflow quedaría justificado si entrenáramos distribuido con GPUs — y la GPU quedó descartada por escrito en 04-05.

Conceptos de Prefect: flow, task, deployment, schedule, work pool

Prefect tiene más piezas de las que necesitamos; estas cinco bastan para todo lo que haremos:

  • Task: una función Python decorada con @task. Es la unidad de reintento, timeout y logging. Regla práctica: una task = un paso que querrías reintentar o ver en la UI por separado.
  • Flow: una función decorada con @flow que llama a tasks (y opcionalmente a otros flows). Define el orden y las dependencias — implícitas: si la task B recibe el resultado de la A, B espera a A.
  • Deployment: un flow registrado en el servidor de Prefect con su configuración: desde dónde se ejecuta, con qué parámetros y con qué schedule. Es lo que convierte "una función en mi repo" en "algo que el servidor sabe lanzar".
  • Schedule: el calendario del deployment; usaremos una expresión cron.
  • Work pool y worker: el work pool es la cola donde el servidor deja trabajo pendiente; el worker es el proceso (en nuestro caso, tipo process: un simple proceso Python en un pod) que escucha esa cola y ejecuta los flows. Esta separación permite que el servidor no ejecute nada él mismo: solo coordina.
flowchart LR
    S[Servidor Prefect<br/>schedules + estado + UI] -->|"lunes 06:00: hay trabajo"| WP[(Work pool<br/>cineclick-pool)]
    WP -->|lo recoge| W[Worker 'process'<br/>pod en el cluster]
    W -->|ejecuta| F[flow scoring_semanal]
    F -->|estados y logs| S

El flow real: flows/scoring_semanal.py

Traducimos la decisión de 04-01 (scoring batch semanal para la campaña de retención) a un flow con seis tasks. Cada task reutiliza piezas que ya existen: el dvc pull del módulo 2, el esquema_clientes de pandera de 05-01, el champion del registry del módulo 3 y la predicción vectorizada de 04-05.

# flows/scoring_semanal.py
"""Scoring batch semanal de churn para la campana de retencion.
Programado: lunes 06:00. Salida: data/predicciones_churn.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 construir_features
from cineclick_churn.validacion import esquema_clientes

RUTA_DATOS = "data/clientes_churn.csv"
RUTA_SALIDA = "data/predicciones_churn.parquet"
URI_CHAMPION = "models:/churn-cineclick@champion"
WEBHOOK_MARKETING = "https://hooks.slack.com/services/XXX/simulado"


@task(retries=3, retry_delay_seconds=[30, 120, 300], timeout_seconds=600)
def descargar_datos_frescos() -> pd.DataFrame:
    """dvc pull del dataset. Fragil por red: 3 reintentos con backoff."""
    subprocess.run(["dvc", "pull", RUTA_DATOS], check=True)
    return pd.read_csv(RUTA_DATOS)


@task  # sin retries: si los datos son invalidos, reintentar no los arregla
def validar_datos(df: pd.DataFrame) -> pd.DataFrame:
    """El mismo esquema pandera de la CI, ahora sobre los datos completos."""
    return esquema_clientes.validate(df, lazy=True)


@task(retries=2, retry_delay_seconds=60, timeout_seconds=300)
def cargar_champion():
    """Carga models:/churn-cineclick@champion. Fragil por red (registry)."""
    return mlflow.sklearn.load_model(URI_CHAMPION)


@task
def predecir_lote(modelo, df: pd.DataFrame) -> pd.DataFrame:
    """Prediccion vectorizada sobre todo el dataframe de una vez
    (la leccion 04-05: nada de iterar filas, ~370x mas rapido)."""
    X = construir_features(df.drop(columns=["abandono"], errors="ignore"))
    probas = modelo.predict_proba(X)[:, 1]
    return pd.DataFrame({
        "id_cliente": df["id_cliente"],
        "probabilidad_abandono": probas,
        "fecha_scoring": datetime.now().date().isoformat(),
    })


@task
def escribir_predicciones(predicciones: pd.DataFrame) -> str:
    predicciones.to_parquet(RUTA_SALIDA, index=False)
    return RUTA_SALIDA


@task(retries=3, retry_delay_seconds=30)
def notificar_marketing(ruta: str, n_clientes: int, n_riesgo: int) -> None:
    """Webhook (Slack simulado): marketing sabe que el fichero esta listo."""
    requests.post(WEBHOOK_MARKETING, json={
        "text": (f"Scoring semanal de churn listo: {ruta} — "
                 f"{n_clientes} clientes, {n_riesgo} en riesgo (p >= 0.5).")
    }, timeout=10).raise_for_status()


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

    df = descargar_datos_frescos()
    df = validar_datos(df)
    modelo = cargar_champion()
    predicciones = predecir_lote(modelo, df)
    ruta = escribir_predicciones(predicciones)

    n_riesgo = int((predicciones["probabilidad_abandono"] >= 0.5).sum())
    logger.info("Scoring completado: %s filas, %s en riesgo", len(predicciones), n_riesgo)
    notificar_marketing(ruta, len(predicciones), n_riesgo)


if __name__ == "__main__":
    scoring_semanal()   # ejecutable en local tal cual, sin servidor

Puntos que merecen pararse:

  • El flow se lee como el proceso: descargar → validar → cargar champion → predecir → escribir → notificar. Las dependencias no se declaran en YAML: salen de que cada task consume el retorno de la anterior. Si validar_datos falla, nada posterior se ejecuta.
  • Los reintentos no son uniformes, y eso es deliberado. descargar_datos_frescos y cargar_champion fallan por red — reintentar tiene sentido. validar_datos falla porque los datos están mal — reintentar la misma validación sobre los mismos datos es perder tiempo y enterrar la alerta; debe fallar rápido y alto.
  • Reutilización total: construir_features y esquema_clientes son exactamente los mismos objetos que usan el API online, la CI y el entrenamiento. El scoring batch no tiene su propia lógica de features ni su propio contrato de datos — esa fue la batalla ganada en el módulo 3.
  • Ejecutable en local: python flows/scoring_semanal.py corre el flow completo sin servidor ni worker. Depurar un flow de Prefect es depurar una función Python — una de las razones de la elección.

Fallos: retries con backoff, timeout y alertas

Anatomía de los reintentos

@task(retries=3, retry_delay_seconds=[30, 120, 300], timeout_seconds=600)
  • retries=3: hasta 4 intentos en total (el original más 3).
  • retry_delay_seconds=[30, 120, 300]: espera creciente (backoff) — 30 s, luego 2 min, luego 5 min. Un fallo transitorio de red suele resolverse solo; martillear cada segundo solo añade carga al servicio que ya está sufriendo.
  • timeout_seconds=600: si el dvc pull se queda colgado (peor que fallar: no avisa), a los 10 minutos Prefect mata el intento y cuenta como fallo — que a su vez dispara el siguiente reintento.

Y cuando falla de verdad

Supón que el lunes el remote almacen está caído una hora entera. Los 4 intentos se agotan (06:00, 06:10 con timeout + esperas…), la task queda en estado Failed, el flow entero queda Failed, y ocurren tres cosas:

  1. Estado visible: la ejecución aparece en rojo en la UI de Prefect, con el traceback de cada intento y sus tiempos. El martes no hay que reconstruir qué pasó — está escrito.
  2. Alerta: una automation de Prefect ("cuando un flow run de scoring-semanal-churn entre en Failed → notifica al canal #alertas-ml") avisa al equipo. La configuración es declarativa en el servidor; no hay que programar nada en el flow. Distíngela de notificar_marketing: esa task es parte del proceso de negocio (avisar de que el parquet está listo); la automation es operación (avisar de que algo se rompió).
  3. Reejecución manual barata: arreglado el remote, un clic en "Retry" en la UI (o prefect flow-run retry <id>) relanza la ejecución. No hace falta esperar al lunes siguiente ni acordarse de los argumentos.

La consecuencia para marketing también está controlada: como la notificación solo se envía al final, no reciben aviso de un parquet que no existe. Sin noticia = sin fichero nuevo; con la alerta interna, el equipo puede avisar a marketing del retraso proactivamente.

Programar el lunes a las 06:00

Con Prefect 2/3, el deployment se declara desde el propio flow y se sirve con un worker. Primero, el work pool (una vez):

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

Después, registrar el deployment con su schedule — la expresión cron 0 6 * * 1 se lee: minuto 0, hora 6, cualquier día del mes, cualquier mes, día de la semana 1 (lunes):

# flows/desplegar_flows.py
from scoring_semanal import scoring_semanal

if __name__ == "__main__":
    scoring_semanal.deploy(
        name="scoring-semanal-lunes",
        work_pool_name="cineclick-pool",
        cron="0 6 * * 1",              # lunes 06:00
        # la zona horaria del schedule se fija a Europe/Madrid en el
        # servidor; sin ella, "06:00" seria UTC y en verano llegaria a las 08:00
    )

Y un worker escuchando el pool (en producción, un pod pequeño del clúster con el repo instalado; en desarrollo, tu terminal):

prefect worker start --pool cineclick-pool

Con esto, la decisión tomada en 04-01 — "scoring semanal, lunes a las 06:00" — por fin se cumple sola: el servidor crea la ejecución a la hora programada, el worker la recoge, y si nadie está mirando… precisamente para eso están los estados y las alertas.

Esbozo del segundo flow: reentrenamiento

El scoring no es el único proceso recurrente en el horizonte: tarde o temprano el modelo habrá que reentrenarlo con datos frescos. El flow es estructuralmente parecido y reutiliza aún más — envuelve el pipeline DVC del módulo 2 entero:

# flows/reentrenamiento.py — ESBOZO, se completa en el modulo 6
from prefect import flow, task
import subprocess


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


@task(timeout_seconds=3600)
def reproducir_pipeline():
    """dvc repro ejecuta preparar_datos -> construir_features ->
    entrenar -> evaluar, con su cache de stages intacta."""
    subprocess.run(["dvc", "repro"], check=True)


@task
def registrar_challenger():
    """Lee metrics.json; si supera los umbrales de 05-01, registra la
    nueva version en MLflow y le pone el alias @challenger.
    NUNCA @champion: la promocion sigue el flujo humano de 05-02."""
    ...


@flow(name="reentrenamiento-churn")
def reentrenamiento():
    datos_frescos()
    reproducir_pipeline()
    registrar_challenger()

Fíjate en lo que este esbozo decide y en lo que no decide. Decide el cómo: DVC hace el trabajo pesado (con su caché de stages: si los datos no cambiaron, preparar_datos ni se ejecuta), y el resultado llega como máximo a @challenger — la promoción a @champion mantiene la aprobación humana de 05-02, también cuando el entrenamiento es automático. Lo que no decide es el cuándo: ¿cada mes? ¿cuando la deriva de datos supere un umbral? ¿cuando el rendimiento caiga? Esa política es materia de la lección 06-03, cuando tengamos la monitorización que la informa. De momento, el flow existe y puede lanzarse a mano desde la UI — que ya es infinitamente mejor que "Laura lo lanza desde su portátil".

Orquestador vs. CI/CD: qué va en cada sitio

Con tres herramientas de automatización en juego (CI, CD, orquestador) es fácil meter las cosas donde no tocan. El criterio de reparto es el disparador:

Proceso Disparador Herramienta Por qué
Tests de código/datos/modelo Push, PR GitHub Actions (CI) Evento de código; efímero; por commit
Build y despliegue de la imagen Tag v*.*.* GitHub Actions (CD) Evento de código; entrega de software
Scoring batch semanal Calendario (lunes 06:00) Prefect Proceso de datos recurrente; necesita retries, estado y alertas
Reentrenamiento Calendario o señal de monitorización (06-03) Prefect Proceso de datos largo, con dependencias y reintentos
Promoción de modelo a @champion Decisión humana Script + revisión (05-02) No es ni evento de código ni calendario: es una decisión

La regla mnemotécnica: GitHub Actions responde a "el código cambió"; Prefect responde a "toca procesar datos". Los runners de Actions son efímeros y anónimos — perfectos para tests, pésimos para un proceso de negocio cuyo historial de ejecuciones quieres consultar dentro de tres meses. Podrías forzar un schedule: en Actions para el scoring (existe), pero perderías reintentos por paso, timeout por paso, reejecución selectiva y la UI de estados — es decir, todo lo que pedimos al principio de la lección.

Errores Comunes y Consejos

  • Un flow de una sola task gigante. Si scoring_semanal fuera una única task que lo hace todo, un fallo del webhook al final obligaría a repetir también la descarga y la predicción. La granularidad de las tasks define la granularidad de los reintentos: separa lo frágil (red) de lo determinista (cálculo).
  • Retries en todo "por si acaso". Reintentar una validación de datos fallida retrasa la alerta 10 minutos para llegar al mismo error. Reintenta lo transitorio (red, servicios externos); falla rápido en lo determinista.
  • Olvidar la zona horaria del schedule. 0 6 * * 1 en UTC son las 07:00 u 08:00 en Madrid según la época del año — y marketing esperando desde las 06:00. Fija explícitamente Europe/Madrid en el schedule del deployment.
  • Lógica de negocio dentro del flow en vez del paquete. Si predecir_lote reimplementara las features "porque era más cómodo", tendríamos la tercera copia de la lógica que el módulo 3 unificó. Las tasks orquestan; el paquete cineclick_churn implementa.
  • Confundir la notificación de negocio con la alerta de fallo. notificar_marketing avisa del éxito a un consumidor; la automation de Prefect avisa del fallo al equipo. Si mezclas ambas en la misma task, un fallo del scoring puede acabar notificando "fichero listo" a marketing o dejando al equipo sin alerta.
  • No probar el flow en local antes de desplegarlo. El flow es una función: python flows/scoring_semanal.py lo ejecuta entero. Depurar mirando logs del worker en el servidor es la manera lenta.
  • Consejo: dale al flow un nombre estable y descriptivo (scoring-semanal-churn) desde el primer día. Las automations, los dashboards y las conversaciones de incidentes se cuelgan de ese nombre; renombrarlo después rompe más de lo que parece.

Ejercicios

Ejercicio 1

Marketing pide que, además del parquet, el scoring escriba un CSV con solo los 500 clientes de mayor probabilidad de abandono (data/top_riesgo.csv), y que si este paso extra falla, el flow se considere igualmente exitoso (el parquet es lo crítico; el CSV es cortesía). Escribe la task y su integración en el flow. Pista: investiga return_state=True o el manejo de excepciones dentro del flow.

Ejercicio 2

El lunes a las 06:00 el servidor de MLflow está caído por mantenimiento no avisado y vuelve a las 06:20. Con el flow tal y como está escrito, ¿el scoring del lunes se completa o falla? Justifícalo calculando la línea temporal de los intentos de cargar_champion (retries=2, retry_delay_seconds=60, timeout_seconds=300) suponiendo que cada intento contra el servidor caído falla a los ~10 segundos y que las tasks previas terminaron a las 06:05.

Ejercicio 3

Un compañero propone mover la CI al orquestador: "así lo tenemos todo en Prefect, un flow ci que corre ruff y pytest cuando alguien haga push". Da tres razones concretas por las que los tests de la lección 05-01 encajan mejor en GitHub Actions que en Prefect.

Soluciones

Ejercicio 1

@task(retries=1, retry_delay_seconds=30)
def escribir_top_riesgo(predicciones: pd.DataFrame, n: int = 500) -> str:
    top = predicciones.nlargest(n, "probabilidad_abandono")
    top.to_csv("data/top_riesgo.csv", index=False)
    return "data/top_riesgo.csv"

Y en el flow, después de escribir_predicciones:

    estado = escribir_top_riesgo(predicciones, return_state=True)
    if estado.is_failed():
        logger.warning("top_riesgo.csv no generado; el parquet SI esta disponible")

Con return_state=True la llamada devuelve el estado en lugar de propagar la excepción: la task aparece en rojo en la UI (visibilidad intacta) pero el flow continúa y termina en éxito. Es la forma idiomática de marcar un paso como "deseable pero no crítico". La alternativa de un try/except alrededor funcionaría, pero el estado del flujo quedaría menos fiel en la UI.

Ejercicio 2

Se completa. Línea temporal: tasks previas listas a las 06:05. Intento 1 de cargar_champion ≈ 06:05:00, falla a los ~10 s (06:05:10). Espera 60 s → intento 2 ≈ 06:06:10, falla (06:06:20). Espera 60 s → intento 3 (el último: retries=2 = 3 intentos en total) ≈ 06:07:20, falla (06:07:30). Resultado: los tres intentos se agotan a las 06:07:30 y el servidor no vuelve hasta las 06:20 — el flow falla, la automation alerta, y a las 06:20+ alguien (o un clic en Retry) lo relanza. Moraleja doble: (a) calcula la ventana total que cubren tus reintentos (aquí ~2,5 minutos, insuficiente para un mantenimiento de 20); para dependencias con caídas largas conocidas, un backoff más largo ([60, 300, 900]) cubre ~21 minutos; (b) aun cuando los reintentos no bastan, el sistema degradó bien: estado visible, alerta emitida, reejecución de un clic — comparadlo con el cron silencioso del inicio.

Ejercicio 3

(1) Disparador: la CI reacciona a push/PR, un evento que GitHub emite de forma nativa a Actions; llevarlo a Prefect exigiría montar webhooks y perdería la integración con el flujo de revisión. (2) Bloqueo del merge: el valor de la CI es marcar la PR en verde/rojo y bloquear el merge (branch protection); Prefect no participa en ese mecanismo — tendrías tests que corren pero no protegen nada. (3) Entorno efímero por commit: cada run de Actions parte de un runner limpio y hace checkout del commit exacto de la PR, que es justo lo que quieres para validar código; el worker de Prefect es un proceso persistente con una versión del repo instalada, pensado para procesos de datos, no para probar N ramas concurrentes. Es la tabla de la lección en acción: evento de código → Actions; proceso de datos recurrente → Prefect.

Conclusión

El último proceso manual de CineClick ya se mueve solo: el flow scoring-semanal-churn (en flows/scoring_semanal.py) descarga los datos con dvc pull, los valida con el mismo esquema pandera de la CI, carga models:/churn-cineclick@champion, predice en lote vectorizado, escribe data/predicciones_churn.parquet y avisa a marketing — cada lunes a las 06:00 (0 6 * * 1, Europe/Madrid) vía deployment sobre el work pool cineclick-pool, con reintentos con backoff donde hay red, timeout donde puede haber cuelgues, y alerta al equipo cuando falla de verdad. Elegimos Prefect frente a Airflow, Dagster y Kubeflow por lo que CineClick es: dos pipelines, un equipo pequeño y código Python que ya funcionaba. Y quedó repartido el mapa de la automatización: Actions para eventos de código, Prefect para procesos de datos, y las decisiones — la promoción a @champion, incluso desde el flow de reentrenamiento esbozado — para las personas. El módulo tiene ya CI, CD y orquestación; queda su pieza más delicada. Porque todo lo construido despliega el modelo nuevo de golpe tras staging, y las métricas offline — por buenas que sean — no garantizan lo que hará con tráfico real. La última lección del módulo enseña a soltar un modelo con red: shadow, canary y A/B.

◄ Anterior
◄
Siguiente ►
►
El Proyecto | Sobre nosotros | Contribuir | Donaciones | Licencia
© Copyright 2026. Todos los derechos reservados