Concepto clave

En Apache Airflow, la configuración de retries, timeouts y alertas por email es fundamental para construir pipelines ETL robustos que puedan manejar fallos temporales sin intervención manual. Imagina un sistema de entrega de paquetes: si el primer intento falla por tráfico, el sistema programa reintentos automáticos; si el paquete tarda demasiado, se marca como timeout y se notifica al remitente. De manera similar, en Airflow, los retries permiten que una tarea se reintente automáticamente tras un fallo (como un error de conexión a una base de datos), los timeouts evitan que las tareas se ejecuten indefinidamente (protegiendo recursos), y las alertas por email notifican a los ingenieros cuando ocurren fallos críticos, permitiendo una respuesta rápida.

Estos mecanismos transforman un DAG de "brittle" a resiliente. Sin ellos, un fallo temporal podría detener todo el pipeline, requiriendo intervención manual. Con una configuración adecuada, el sistema se auto-cura para errores comunes y escala en entornos de producción. La clave es equilibrar: demasiados retries pueden enmascarar problemas reales, mientras que timeouts muy cortos pueden causar falsos positivos.

Cómo funciona en la práctica

Configurar retries, timeouts y alertas en Airflow implica parámetros a nivel de DAG y tarea. Aquí un ejemplo paso a paso para un DAG que procesa datos diarios:

Define parámetros globales en el DAG: Usa default_args para establecer valores por defecto que apliquen a todas las tareas.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data_engineer',
    'retries': 3,               # Número de reintentos
    'retry_delay': timedelta(minutes=5),  # Espera entre reintentos
    'email': ['[email protected]'],
    'email_on_failure': True,   # Envía email si falla
    'email_on_retry': False,    # Opcional: notificar en cada reintento
    'execution_timeout': timedelta(hours=2)  # Timeout global
}

dag = DAG(
    'etl_diario',
    default_args=default_args,
    start_date=datetime(2023, 1, 1),
    schedule_interval='@daily'
)

Ajusta parámetros a nivel de tarea: Sobrescribe valores para tareas específicas si es necesario.

def extraer_datos():
    # Código para extraer datos
    pass

extract_task = PythonOperator(
    task_id='extraer',
    python_callable=extraer_datos,
    retries=5,  # Más reintentos para esta tarea crítica
    execution_timeout=timedelta(hours=1),  # Timeout específico
    dag=dag
)

Configura el servidor de email: En airflow.cfg, establece:

[smtp]
smtp_host = smtp.gmail.com
smtp_starttls = True
smtp_ssl = False
smtp_port = 587
smtp_mail_from = [email protected]
smtp_user = tu_usuario
smtp_password = tu_contraseña

Caso de estudio

Una empresa de e-commerce usa un DAG para procesar ventas diarias. El pipeline incluye: extraer datos de una API externa (propensa a timeouts), transformarlos en un data warehouse, y cargar reportes. Sin configuración, fallos en la API detenían el proceso, requiriendo acción manual a las 3 AM.

Solución implementada:

Tarea	Retries	Retry Delay	Timeout	Email on Failure
Extraer de API	5	10 minutos	30 minutos	Sí
Transformar datos	3	5 minutos	2 horas	Sí
Cargar reporte	2	2 minutos	1 hora	No (solo log)

Resultado: En 3 meses, el 95% de los fallos de API se resolvieron automáticamente con retries, reduciendo alertas manuales en un 70%. Los timeouts previenen que tareas de transformación consuman recursos indefinidamente.

Errores comunes

Configurar retries infinitos: Poner retries muy alto (ej., 100) puede ocultar errores persistentes. Mejor práctica: usar 3-5 reintentos y monitorear fallos recurrentes.
Ignorar el retry_delay: Sin delay entre reintentos, se satura el sistema. Usa timedelta para espaciar reintentos (ej., 5-10 minutos).
Timeouts muy cortos en tareas largas: Un execution_timeout de 10 minutos en una tarea que normalmente tarda 15 causará fallos innecesarios. Basa los timeouts en métricas históricas.
No probar alertas de email: Configurar email_on_failure sin verificar que el servidor SMTP funciona lleva a alertas silenciosas. Prueba con un DAG de prueba que falle intencionalmente.
Sobrescribir parámetros incorrectamente: Al definir tareas, asegúrate de que los parámetros como retries se pasen correctamente al operador, no al DAG.

Checklist de dominio

Puedo configurar default_args en un DAG con retries, retry_delay, y email_on_failure.
Sé ajustar retries y timeouts a nivel de tarea para casos específicos.
He probado alertas por email con un fallo controlado en mi entorno.
Entiendo cuándo usar email_on_retry (para debugging) vs. email_on_failure (para alertas críticas).
Puedo interpretar logs de Airflow para identificar si un fallo fue por timeout o agotamiento de retries.
Sé configurar el servidor SMTP en airflow.cfg para entornos de producción.
He equilibrado retries y timeouts basándome en el comportamiento real de mis tareas.

Configura un DAG con manejo robusto de fallos

En este ejercicio, crearás un DAG que simula un pipeline ETL con manejo avanzado de errores. Sigue estos pasos:

Crea un nuevo archivo Python en tu directorio de DAGs de Airflow (ej., ~/airflow/dags/ejercicio_retries.py).
Define el DAG con los siguientes parámetros globales:
- Nombre: ejercicio_manejo_fallos
- Schedule: diario (@daily)
- Retries: 4
- Retry delay: 3 minutos
- Email on failure: verdadero (usa un email ficticio como [email protected])
- Timeout de ejecución: 1 hora
Añade tres tareas usando PythonOperator:
- Tarea 1 (tarea_inestable): Una función que aleatoriamente falla el 50% de las veces (usa random para simular esto). Configúrala con 6 retries y un timeout de 10 minutos.
- Tarea 2 (tarea_larga): Una función que duerme por 45 minutos (usa time.sleep). Configúrala con 2 retries y un timeout de 1 hora.
- Tarea 3 (targa_exitosa): Una función que solo imprime un mensaje. Sin configuración especial.
Establece dependencias: tarea_inestable -> tarea_larga -> tarea_exitosa.
Ejecuta el DAG manualmente desde la UI de Airflow y observa el comportamiento en los logs. ¿Cuántos reintentos ocurren para tarea_inestable? ¿La tarea_larga alcanza el timeout?
Modifica los parámetros: Cambia el timeout de tarea_larga a 30 minutos y vuelve a ejecutar. Documenta lo que sucede.

Pistas

Usa import random y random.random() > 0.5 para simular fallos aleatorios en la tarea inestable.
Para la tarea larga, recuerda que time.sleep(2700) son 45 minutos, pero en pruebas puedes usar valores más cortos como 60 segundos.
Verifica que las tareas usen el operador correcto: PythonOperator(task_id='nombre', python_callable=funcion, ...).

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.

Configuración de Retries, Timeouts y Alertas por Email