Orquesta Pipelines de ML con Prefect
Convierte scripts frágiles de entrenamiento en pipelines de ML robustos y observables con Prefect: reintentos, caché y re-entrenamiento programado.
La mayoría de los workflows de ML comienzan como un solo script: entrenar.py. Funciona bien localmente. Luego alguien pide re-entrenamiento programado. Luego otra persona quiere saber por qué falló la ejecución del martes pasado. Luego necesitas re-entrenar solo cuando llegan datos nuevos. De repente tu script se convirtió en una maraña de cron jobs, shell scripts y alertas de Slack pegada con cinta adhesiva.
Prefect resuelve exactamente este problema. Convierte tus funciones de Python existentes en workflows observables, reintentables y programables — sin que tengas que aprender un nuevo DSL ni reestructurar tu código. En esta guía tomo un script de entrenamiento monolítico y lo convierto en un flow de Prefect con reintentos, caché, re-entrenamiento programado y promoción condicional del modelo.
Lo Que Te Da Prefect
- Tasks: unidades individuales de trabajo, cada una con lógica automática de reintentos
- Flows: composiciones de tasks que Prefect orquesta
- Deployments: ejecuciones programadas o disparadas por eventos de tus flows
- UI: un dashboard integrado para historial de ejecuciones, logs e inspección de fallos
pip install prefect scikit-learn pandas joblib
De Script de Entrenamiento a Flow de Prefect
Aquí hay un script típico de entrenamiento ML antes de Prefect:
# Antes: script monolítico
import pandas as pd
import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score
df = pd.read_csv('datos/churn.csv')
X = df.drop('churn', axis=1)
y = df['churn']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
model = RandomForestClassifier(n_estimators=200, random_state=42)
model.fit(X_train, y_train)
f1 = f1_score(y_test, model.predict(X_test))
print(f"F1: {f1:.4f}")
joblib.dump(model, 'model.joblib')
Si esto falla en el paso 3, no tienes forma de saber dónde falló, no hay lógica de reintentos ni historial. Ahora con Prefect:
# Después: flow de Prefect
import pandas as pd
import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, classification_report
from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta
@task(
retries=3,
retry_delay_seconds=10,
cache_key_fn=task_input_hash,
cache_expiration=timedelta(hours=1),
)
def cargar_datos(path: str) -> pd.DataFrame:
logger = get_run_logger()
logger.info(f"Cargando datos desde {path}")
df = pd.read_csv(path)
logger.info(f"Cargados {len(df)} filas, {df.shape[1]} columnas")
return df
@task
def preprocesar(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.Series]:
logger = get_run_logger()
# Eliminar filas sin target
df = df.dropna(subset=['churn'])
logger.info(f"Después de eliminar nulos: {len(df)} filas")
X = df.drop('churn', axis=1)
y = df['churn'].astype(int)
return X, y
@task
def dividir_datos(X: pd.DataFrame, y: pd.Series, test_size: float = 0.2):
return train_test_split(X, y, test_size=test_size, random_state=42, stratify=y)
@task
def entrenar_modelo(X_train: pd.DataFrame, y_train: pd.Series, n_estimadores: int = 200):
logger = get_run_logger()
logger.info(f"Entrenando RandomForest con {n_estimadores} árboles")
model = RandomForestClassifier(n_estimators=n_estimadores, random_state=42, n_jobs=-1)
model.fit(X_train, y_train)
return model
@task
def evaluar_modelo(model, X_test: pd.DataFrame, y_test: pd.Series) -> dict:
logger = get_run_logger()
y_pred = model.predict(X_test)
f1 = f1_score(y_test, y_pred, average='weighted')
acc = model.score(X_test, y_test)
logger.info(f"F1: {f1:.4f} | Accuracy: {acc:.4f}")
logger.info(f"\n{classification_report(y_test, y_pred)}")
return {'f1': f1, 'accuracy': acc}
@task
def guardar_modelo(model, metricas: dict, output_path: str = 'model.joblib'):
logger = get_run_logger()
joblib.dump({'model': model, 'metricas': metricas}, output_path)
logger.info(f"Modelo guardado en {output_path} (F1={metricas['f1']:.4f})")
@flow(name="pipeline-entrenamiento-churn")
def pipeline_entrenamiento(
data_path: str = 'datos/churn.csv',
n_estimadores: int = 200,
model_output: str = 'model.joblib',
):
df = cargar_datos(data_path)
X, y = preprocesar(df)
X_train, X_test, y_train, y_test = dividir_datos(X, y)
model = entrenar_modelo(X_train, y_train, n_estimadores)
metricas = evaluar_modelo(model, X_test, y_test)
guardar_modelo(model, metricas, model_output)
return metricas
if __name__ == '__main__':
resultado = pipeline_entrenamiento()
print(f"Métricas finales: {resultado}")
Ejecuta directamente:
python pipeline.py
Prefect registra cada task individualmente, muestra su estado y reintenta cargar_datos hasta 3 veces si el archivo no está disponible temporalmente.
Iniciando la UI de Prefect
prefect server start
Abre http://localhost:4200. Verás un dashboard con el historial de ejecuciones de tus flows, su duración, estado a nivel de task y logs — todo sin configuración extra.
Caché de Tasks Costosas
El argumento cache_key_fn=task_input_hash en el decorador de cargar_datos significa que si ejecutas el flow dos veces con el mismo data_path, la segunda ejecución omitirá la carga y usará el resultado en caché. Esto es invaluable durante el desarrollo cuando iteras sobre el código de entrenamiento pero los datos no han cambiado.
# Primera ejecución: carga desde disco (lenta)
pipeline_entrenamiento(data_path='datos/churn.csv')
# Segunda ejecución en menos de 1 hora: usa caché (instantáneo)
pipeline_entrenamiento(data_path='datos/churn.csv')
Re-entrenamiento Programado
Para ejecutar tu flow según un schedule, crea un deployment con serve:
# deploy.py
from pipeline import pipeline_entrenamiento
if __name__ == '__main__':
pipeline_entrenamiento.serve(
name="reentrenamiento-semanal-churn",
cron="0 6 * * 1", # Todos los lunes a las 6 AM UTC
parameters={
"data_path": "datos/churn_mas_reciente.csv",
"n_estimadores": 300,
},
)
python deploy.py
El comando serve mantiene un proceso ligero vivo que espera el próximo trigger programado. También puedes dispararlo manualmente desde la UI o CLI:
prefect deployment run 'pipeline-entrenamiento-churn/reentrenamiento-semanal-churn'
Re-entrenamiento Condicional Basado en Métricas
Un patrón común es promover un nuevo modelo solo si supera al actual:
@task
def cargar_metricas_actuales(model_path: str = 'model.joblib') -> dict:
import os
if not os.path.exists(model_path):
return {'f1': 0.0}
guardado = joblib.load(model_path)
return guardado.get('metricas', {'f1': 0.0})
@flow(name="reentrenamiento-condicional")
def pipeline_condicional(data_path: str = 'datos/churn.csv'):
logger = get_run_logger()
df = cargar_datos(data_path)
X, y = preprocesar(df)
X_train, X_test, y_train, y_test = dividir_datos(X, y)
model = entrenar_modelo(X_train, y_train)
nuevas_metricas = evaluar_modelo(model, X_test, y_test)
metricas_actuales = cargar_metricas_actuales()
if nuevas_metricas['f1'] > metricas_actuales['f1'] + 0.005:
guardar_modelo(model, nuevas_metricas)
logger.info(f"Modelo promovido: {nuevas_metricas['f1']:.4f} > {metricas_actuales['f1']:.4f}")
else:
logger.info(f"Modelo NO promovido: {nuevas_metricas['f1']:.4f} <= {metricas_actuales['f1']:.4f}")
return nuevas_metricas
Si versionas tus modelos con MLflow, este paso de promoción se corresponde de forma natural con los stages del model registry — mira Gestiona el Ciclo de Vida de Machine Learning con MLflow en Python.
Manejo de Fallos con Hooks de Estado
Prefect ofrece manejo de fallos de primera clase mediante reintentos, notificaciones y hooks de estado:
from prefect import flow
def al_fallar(flow, flow_run, state):
# Enviar alerta: Slack, email, PagerDuty...
print(f"Flow {flow.name} falló: {state.message}")
@flow(
name="pipeline-monitoreado",
on_failure=[al_fallar],
)
def pipeline_monitoreado():
# ...
pass
Conclusión
Prefect transforma un script Python ad-hoc en un workflow de ML listo para producción en pocas docenas de líneas. El mismo código que corre localmente corre en producción, con reintentos, caché, programación y una UI para inspeccionar cada ejecución. A medida que tus sistemas de ML maduran, la inversión en orquestación se paga sola cada vez que necesitas depurar por qué un modelo se comportó de forma inesperada — o por qué, sin que nadie lo notara, no se re-entrenó cuando debería haberlo hecho.
Cuando el re-entrenamiento ya corre de forma confiable, lo siguiente es saber cuándo un modelo ha comenzado a degradarse: Monitoreo de Modelos ML en Producción con Evidently.
Sobre el autor
Rodrigo Arenas es arquitecto de software e ingeniero de machine learning en Medellín, Colombia. Construye productos, plataformas y software open source para IA, Data Engineering y Machine Learning — es el creador de Ciaren, sklearn-genetic-opt y PyWorkforce.
Lectura recomendada
Rodrigo Arenas — ML Engineer & Open Source Maintainer
¿Estás trabajando en algo similar? Puedo ayudarte a llevarlo a producción.