Autoría de DAGs de Airflow

¿Qué hace Autoría de DAGs de Airflow?

Cuándo usar

• Crear o refactorizar DAGs y tareas de Airflow.
• Configurar schedules, catchup/backfill, reintentos y SLAs.
• Pasar datos entre tareas (XCom) o usar conexiones/variables.
• NO usar para diagnosticar un DAG en ejecución que está roto (usar depuración-de-pipelines-de-airflow).

Flujo de trabajo

- [ ] Hacer que cada tarea sea idempotente y parametrizada por el intervalo de datos
- [ ] Mantener el código costoso/pesado en importaciones dentro de las tareas, no al nivel superior del módulo
- [ ] Configurar schedule + catchup de manera deliberada
- [ ] Configurar reintentos, retry_delay y SLAs
- [ ] Cablear dependencias mediante valores de retorno de TaskFlow o operadores >>

1. Tareas idempotentes — una tarea para el intervalo 2026-01-15 debe producir el mismo resultado sin importar si se ejecuta una vez o se vuelve a ejecutar. Usa el intervalo de datos, no datetime.now().
2. Sin código pesado de nivel superior — el scheduler analiza cada archivo DAG con frecuencia; las llamadas a bases de datos, API o importaciones grandes a nivel de módulo ralentizan la programación y pueden romper el análisis. Colócalas dentro de las tareas.
3. Schedule + catchup a propósito — catchup=True rellena cada intervalo perdido desde start_date; por defecto pon False a menos que quieras eso.
4. Reintentos y SLAs — los fallos transitorios son normales; establece retries y retry_delay; usa SLAs/alertas para la demora.

Patrones

DAG TaskFlow, idempotente y correctamente cableado:

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["orders"],
)
def orders_pipeline():

    @task
    def extract(data_interval_start=None):
        # Usa el intervalo, no now(), para que las re-ejecuciones sean deterministas.
        return fetch_orders(day=data_interval_start.date())

    @task
    def load(rows):
        # Borra-inserta la partición -> idempotente al reintentar.
        overwrite_partition("fct_orders", rows)

    load(extract())

orders_pipeline()

Pasar datos pequeños vía XCom (valores de retorno); pasar datos grandes vía almacenamiento — escribe en S3/GCS/warehouse y pasa la ruta/clave, nunca megabytes a través de XCom.

Usar conexiones/variables para secretos y configuración (BaseHook.get_connection, Variable.get), nunca credenciales hardcodeadas.

Errores comunes

• Llamadas a API/DB de nivel superior o importaciones pesadas — ralentizan el scheduler y pueden fallar el análisis del DAG en todo el despliegue.
• datetime.now() dentro de las tareas — rompe la idempotencia y el backfill; usa data_interval_start/_end.
• catchup=True no intencionado — inunda el clúster con ejecuciones históricas en el primer despliegue.
• Payloads grandes a través de XCom — infla la base de datos de metadatos; pasa referencias.
• start_date dinámico (ej. days_ago) — hace que los schedules sean no deterministas; usa una marca de tiempo fija.
• Una tarea monstruosa — divide extracción/transformación/carga para que los reintentos sean granulares.

Referencias

• Referencia de scheduling, catchup y backfill