Orquestrador de DAGs de Airflow

Orquestador de DAGs de Airflow

Diseña y opera DAGs de Apache Airflow para orquestación confiable de pipelines de datos con gestión adecuada de dependencias, SLAs y monitoreo.

Disparadores de Activación

Activar en: "Airflow", "DAG", "operator", "sensor", "scheduler", "task dependency", "SLA", "backfill", "XCom", "TaskFlow API", "MWAA", "Cloud Composer"

NO para: ejecución de modelos dbt → dbt-analytics-engineer (aunque Airflow puede ejecutar dbt) | Procesamiento de streams → streaming-pipeline-architect | Motores de flujo de trabajo (Temporal) → distributed-transaction-manager

Inicio Rápido

1. Define el DAG — usa TaskFlow API (decoradores @dag, @task) para DAGs nativos en Python
2. Establece el schedule — cron o timetable, con catchup=False a menos que el backfill sea intencional
3. Configura reintentos — retries=2, retry_delay=timedelta(minutes=5) en cada tarea
4. Agrega SLAs — sla=timedelta(hours=2) en tareas de ruta crítica
5. Prueba localmente — airflow dags test my_dag 2026-01-01 antes de desplegar

Capacidades Principales

Dominio	Tecnologías
Airflow	Apache Airflow 2.10+, MWAA, Cloud Composer 3
Operadores	BashOperator, PythonOperator, KubernetesPodOperator
Proveedores	apache-airflow-providers-{snowflake, google, aws, dbt-cloud}
Ejecutores	CeleryExecutor, KubernetesExecutor, LocalExecutor
Monitoreo	Omisiones de SLA, duración de tareas, métricas de Airflow → Prometheus

Patrones de Arquitectura

DAG con TaskFlow API

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule="0 6 * * *",          # diario a las 6am UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(hours=2),
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task()
    def extract_payments() -> dict:
        """Extrae de Stripe API"""
        data = stripe_client.list_payments(date=today())
        return {"count": len(data), "path": "s3://raw/payments/"}

    @task()
    def extract_orders() -> dict:
        """Extrae de Shopify API"""
        data = shopify_client.list_orders(date=today())
        return {"count": len(data), "path": "s3://raw/orders/"}

    @task()
    def transform(payments: dict, orders: dict) -> str:
        """Junta y transforma en DuckDB"""
        result_path = run_duckdb_transform(payments["path"], orders["path"])
        return result_path

    @task()
    def load(path: str):
        """Carga a Snowflake"""
        snowflake_copy_into("fct_revenue", path)

    # Define dependencias mediante llamadas a funciones
    payments = extract_payments()
    orders = extract_orders()
    transformed = transform(payments, orders)
    load(transformed)

daily_revenue_pipeline()

Mapeo Dinámico de Tareas (Fan-Out/Fan-In)

@task()
def get_partitions() -> list[str]:
    return ["2026-01-01", "2026-01-02", "2026-01-03"]

@task()
def process_partition(partition_date: str) -> dict:
    """Se ejecuta en paralelo para cada partición"""
    return {"date": partition_date, "rows": process(partition_date)}

@task()
def aggregate(results: list[dict]):
    """Fan-in: recibe todos los resultados de las particiones"""
    total = sum(r["rows"] for r in results)
    log.info(f"Procesadas {total} filas en total")

# Mapea dinámicamente process_partition en todas las particiones
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)

Integración dbt + Airflow

from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# Opción 1: Cosmos (recomendado)
dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/airflow/dbt/"),
    profile_config=ProfileConfig(
        profile_name="default",
        target_name="prod",
    ),
    schedule="@daily",
    dag_id="dbt_daily",
)

# Opción 2: BashOperator (simple)
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)

Antipatrones

1. Tareas pesadas — las tareas deben ser unidades atómicas; no pongas extraer+transformar+cargar en una sola tarea
2. XCom para datos grandes — XCom es para metadatos (rutas, conteos); pasa datos via S3/GCS, no mediante serialización de XCom
3. catchup=True por accidente — a menos que quieras backfill, establece catchup=False; de lo contrario Airflow ejecutará cada intervalo perdido
4. Sin reintentos — las fallas transitorias son comunes; siempre configura retries >= 1 con un retardo
5. Código de nivel superior en archivos DAG — los archivos DAG se analizan cada 30s; las importaciones pesadas o llamadas a API a nivel de módulo ralentizan el scheduler

Lista de Verificación de Calidad

• Se usa TaskFlow API para DAGs nativos en Python (no encadenamiento de operadores heredados)
• catchup=False a menos que el backfill sea intencional
• Reintentos configurados en todas las tareas (mínimo 1 reintento)
• SLA establecido en tareas de ruta crítica
• XCom pasa referencias (rutas S3), no cargas de datos
• Se usa mapeo dinámico de tareas para paralelismo fan-out
• DAG probado localmente con airflow dags test antes del despliegue
• Idempotencia de tareas: re-ejecutar una tarea produce el mismo resultado
• Alertas configuradas para fallos de tareas y omisiones de SLA
• Etiquetas de DAG aplicadas para filtrado en la interfaz de Airflow