Práctica: Construir un DAG con Múltiples Tareas y Dependencias

Introducción al Diseño de DAGs Complejos

En lecciones anteriores, has aprendido los fundamentos de un DAG (Grafo Acíclico Dirigido) y cómo crear tareas simples. Sin embargo, el verdadero poder de Apache Airflow se despliega cuando orquestamos pipelines complejos con múltiples tareas que tienen relaciones de dependencia intrincadas. En esta práctica, pasaremos de la teoría a la construcción de un DAG que simula un escenario realista de procesamiento de datos, donde el orden, el manejo de errores y la claridad del flujo son críticos.

Imagina que necesitas procesar datos de ventas diarias. El proceso no es una simple secuencia lineal; implica extraer datos de distintas fuentes, validarlos, transformarlos de manera independiente y, finalmente, consolidarlos en un reporte. Diseñar este flujo de manera efectiva requiere una planificación cuidadosa de las dependencias. Un DAG bien estructurado no solo ejecuta las tareas en el orden correcto, sino que también maximiza la eficiencia al permitir la ejecución paralela de tareas independientes y proporciona una visibilidad clara del estado del proceso en su conjunto.

El objetivo de esta lección es guiarte en la construcción de un DAG desde cero que incorpore múltiples operadores, dependencias no lineales (ramificaciones y uniones), y buenas prácticas de diseño. Trabajaremos con un caso concreto: un pipeline ETL para un sistema de recomendación que procesa datos de usuarios, productos y transacciones.

Concepto Clave: El Grafo de Dependencias como un Plan de Vuelo

Piensa en un DAG no como un simple script, sino como un plan de vuelo para tus datos. Cada tarea es una etapa del viaje (despegue, crucero, aterrizaje, descarga de equipaje). Algunas etapas pueden ocurrir en paralelo (descargar el equipaje de la bodega frontal y trasera), pero tienen dependencias estrictas: no puedes comenzar a descargar el equipaje hasta que el avión haya aterrizado y esté en la puerta de embarque. Del mismo modo, no puedes combinar información de usuarios y productos (una tarea de unión) hasta que ambas fuentes de datos hayan sido extraídas y limpiadas de manera independiente.

La magia de Airflow reside en su scheduler, que actúa como el controlador de tráfico aéreo. Tú defines el plan (el DAG), y el scheduler se asegura de que cada tarea (vuelo/etapa) se ejecute solo cuando todas sus dependencias previas (las tareas upstream) se hayan completado con éxito. Esta abstracción te libera de tener que escribir lógica compleja de manejo de estados y errores, permitiéndote centrarte en la lógica de negocio de cada tarea individual.

Tip Clave: Antes de escribir una sola línea de código, dibuja tu DAG en un papel o pizarra. Identifica las tareas que pueden ejecutarse en paralelo y los puntos donde los flujos deben converger. Este diagrama será tu mapa de ruta y te evitará crear dependencias circulares o innecesarias.

Cómo Funciona en la Práctica: Paso a Paso en Nuestro Pipeline

Vamos a construir un DAG para un sistema de recomendación. Nuestro pipeline debe: 1) Extraer datos crudos de tres fuentes (una base de datos de usuarios, un archivo CSV de productos y una API de transacciones). 2) Transformar/limpiar cada conjunto de datos por separado. 3) Validar la calidad de los datos transformados. 4) Unir todos los datos limpios en un dataset único. 5) Entrenar un modelo simple de recomendación. 6) Generar un reporte de salida. 7) Notificar el éxito o fracaso del proceso.

El flujo no es secuencial. Los pasos de extracción y transformación para usuarios, productos y transacciones son independientes entre sí y, por lo tanto, pueden (y deben) ejecutarse en paralelo para ahorrar tiempo. Sin embargo, la tarea de unión (join) depende absolutamente de que las tres tareas de validación previas hayan terminado con éxito. Asimismo, el entrenamiento del modelo depende de la unión, y la generación del reporte depende del entrenamiento. La notificación es una tarea especial que debe ejecutarse al final del proceso, sin importar si algunas tareas fallaron (siempre que se maneje adecuadamente).

Para implementar esto en Airflow, definiremos cada paso como una tarea usando operadores como PythonOperator para la lógica personalizada y EmailOperator para la notificación. Luego, usaremos los operadores de bitshifting (>> y <<) o los métodos set_upstream/set_downstream para establecer explícitamente estas relaciones de dependencia, creando un grafo que el scheduler pueda interpretar y ejecutar.

Código en Acción: DAG Completo para el Sistema de Recomendación

A continuación, se presenta el código completo del DAG. Presta atención a cómo se estructuran las dependencias para permitir el paralelismo y garantizar el orden correcto en los puntos de convergencia.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.operators.email import EmailOperator
from airflow.utils.dates import days_ago

# Argumentos por defecto para el DAG
default_args = {
    'owner': 'equipo_datos',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['[email protected]'],
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Definir el DAG
with DAG(
    'pipeline_recomendacion_complejo',
    default_args=default_args,
    description='Un pipeline ETL complejo para un sistema de recomendación',
    schedule_interval=timedelta(days=1),
    start_date=days_ago(2),
    tags=['ETL', 'recomendacion', 'procesamiento'],
) as dag:

    # 1. Tarea de inicio (marcador)
    inicio = DummyOperator(task_id='inicio_pipeline')

    # 2. Tareas de Extracción (Paralelas)
    def extraer_datos_usuarios():
        # Simulación: Lógica para conectar a BD y extraer
        print("Extrayendo datos de usuarios...")
        return "datos_usuarios_crudos"

    def extraer_datos_productos():
        # Simulación: Leer archivo CSV
        print("Extrayendo datos de productos...")
        return "datos_productos_crudos"

    def extraer_datos_transacciones():
        # Simulación: Llamar a una API
        print("Extrayendo datos de transacciones...")
        return "datos_transacciones_crudas"

    extraer_usuarios = PythonOperator(
        task_id='extraer_usuarios',
        python_callable=extraer_datos_usuarios
    )

    extraer_productos = PythonOperator(
        task_id='extraer_productos',
        python_callable=extraer_datos_productos
    )

    extraer_transacciones = PythonOperator(
        task_id='extraer_transacciones',
        python_callable=extraer_datos_transacciones
    )

    # 3. Tareas de Transformación (Paralelas, cada una depende de su extracción)
    def transformar_usuarios(**context):
        datos = context['task_instance'].xcom_pull(task_ids='extraer_usuarios')
        print(f"Transformando datos de usuarios: {datos}")
        return "usuarios_limpios"

    def transformar_productos(**context):
        datos = context['task_instance'].xcom_pull(task_ids='extraer_productos')
        print(f"Transformando datos de productos: {datos}")
        return "productos_limpios"

    def transformar_transacciones(**context):
        datos = context['task_instance'].xcom_pull(task_ids='extraer_transacciones')
        print(f"Transformando datos de transacciones: {datos}")
        return "transacciones_limpias"

    transformar_usuarios = PythonOperator(
        task_id='transformar_usuarios',
        python_callable=transformar_usuarios,
        provide_context=True
    )

    transformar_productos = PythonOperator(
        task_id='transformar_productos',
        python_callable=transformar_productos,
        provide_context=True
    )

    transformar_transacciones = PythonOperator(
        task_id='transformar_transacciones',
        python_callable=transformar_transacciones,
        provide_context=True
    )

    # 4. Tareas de Validación (Paralelas, cada una depende de su transformación)
    validar_usuarios = DummyOperator(task_id='validar_usuarios')
    validar_productos = DummyOperator(task_id='validar_productos')
    validar_transacciones = DummyOperator(task_id='validar_transacciones')

    # 5. Tarea de Unión (Depende de las TRES validaciones)
    def unir_datasets(**context):
        # Recupera los resultados de las tareas de transformación vía XCom
        u = context['task_instance'].xcom_pull(task_ids='transformar_usuarios')
        p = context['task_instance'].xcom_pull(task_ids='transformar_productos')
        t = context['task_instance'].xcom_pull(task_ids='transformar_transacciones')
        print(f"Uniendo datasets: {u}, {p}, {t}")
        return "dataset_unificado"

    unir_datos = PythonOperator(
        task_id='unir_datos',
        python_callable=unir_datasets,
        provide_context=True
    )

    # 6. Tarea de Entrenamiento del Modelo
    def entrenar_modelo(**context):
        dataset = context['task_instance'].xcom_pull(task_ids='unir_datos')
        print(f"Entrenando modelo con: {dataset}")
        return "modelo_entrenado.pkl"

    entrenar_modelo = PythonOperator(
        task_id='entrenar_modelo',
        python_callable=entrenar_modelo,
        provide_context=True
    )

    # 7. Tarea de Generación de Reporte
    generar_reporte = DummyOperator(task_id='generar_reporte')

    # 8. Tarea de Notificación (se ejecuta siempre al final, éxito o fallo)
    notificar_exito = EmailOperator(
        task_id='notificar_exito',
        to='[email protected]',
        subject='Pipeline de Recomendación Completado - {{ ds }}',
        html_content='El pipeline de recomendación se ha ejecutado exitosamente.'
    )

    # 9. Tarea de fin (marcador)
    fin = DummyOperator(task_id='fin_pipeline')

    # DEFINICIÓN DE LAS DEPENDENCIAS (LA ESTRUCTURA DEL GRAFO)
    inicio >> [extraer_usuarios, extraer_productos, extraer_transacciones]

    extraer_usuarios >> transformar_usuarios >> validar_usuarios
    extraer_productos >> transformar_productos >> validar_productos
    extraer_transacciones >> transformar_transacciones >> validar_transacciones

    [validar_usuarios, validar_productos, validar_transacciones] >> unir_datos

    unir_datos >> entrenar_modelo >> generar_reporte >> notificar_exito >> fin

Este DAG crea un flujo visual claro en la interfaz de Airflow. Verás tres ramas independientes (usuarios, productos, transacciones) que salen de la tarea inicio. Estas ramas convergen en la tarea unir_datos, tras lo cual el flujo continúa de forma lineal hasta el final. La tarea notificar_exito está configurada para ejecutarse tras generar_reporte, pero en un escenario real, podrías querer usar trigger rules como all_done para que se notifique incluso si alguna tarea falla.

Errores Comunes y Cómo Evitarlos

Al diseñar DAGs complejos, es fácil caer en ciertos patrones problemáticos. Aquí te presentamos los más comunes y cómo solucionarlos:

1. Dependencias Circulares (Ciclos en el Grafo): Esto ocurre cuando la tarea A depende de B, B depende de C y C depende de A, formando un bucle. Airflow detectará esto y lanzará un error. Cómo evitarlo: Siempre dibuja tu DAG. Asegúrate de que las flechas de dependencia vayan solo en una dirección a lo largo del tiempo. Usa herramientas de linting como airflow dags list y airflow dags report para detectar ciclos.

2. Paralelismo Innecesario Bloqueado por Dependencias Incorrectas: Si defines por error que transformar_usuarios depende de extraer_productos, estás serializando tareas que podrían ser paralelas, aumentando el tiempo total de ejecución. Cómo evitarlo: Revisa minuciosamente tu definición de dependencias. Agrupa tareas independientes en listas al usar los operadores de bitshift, como se muestra en el código con [validar_usuarios, validar_productos, validar_transacciones].

3. Mal Uso de XCom para Datos Grandes: XCom es ideal para mensajes pequeños (IDs, rutas, estados). Forzarlo a pasar grandes DataFrames o archivos entre tareas saturará la base de datos de metadatos de Airflow. Cómo evitarlo: Para datos voluminosos, usa almacenamiento intermedio como S3, GCS, HDFS o una base de datos compartida. Pasa solo la ruta o la clave del archivo a través de XCom.

4. No Planificar para el Fallo (Trigger Rules por Defecto): La regla por defecto all_success significa que una tarea se ejecuta solo si todas sus tareas ascendentes tienen éxito. Si validar_productos falla, unir_datos nunca se ejecutará, quizás dejando el pipeline a medias. Cómo evitarlo: Evalúa el uso de trigger_rule. Para tareas de limpieza o notificación final, considera all_done (se ejecuta sin importar el estado) o one_success.

Tip de Depuración: Usa la vista Graph o Tree de la interfaz web de Airflow para visualizar tu DAG. Si la estructura no se parece a tu diagrama inicial, hay un error en la definición de dependencias. La vista Task Instance Details es invaluable para rastrear el flujo de XCom y los logs de cada tarea.

Checklist de Dominio

Antes de considerar que dominas el diseño de DAGs avanzados con múltiples dependencias, asegúrate de poder verificar los siguientes puntos:

• Puedo dibujar en un diagrama la secuencia y paralelismo de un proceso de negocio antes de codificarlo.
• Sé definir dependencias entre tareas usando tanto el operador de bitshifting (>>) como el método set_downstream.
• Puedo crear puntos de divergencia (una tara que precede a varias) y puntos de convergencia (varias tareas que preceden a una) en mi DAG.
• Comprendo la diferencia entre las trigger rules comunes (all_success, all_done, one_success) y sé cuándo aplicar cada una.
• Sé cómo pasar información pequeña entre tareas usando XCom y cuándo evitar su uso para datos grandes.
• Puedo identificar y corregir una dependencia circular en un DAG existente.
• Sé cómo utilizar operadores DummyOperator para crear puntos de control y agrupar lógicamente tareas en el grafo.
• Puedo configurar correctamente los argumentos por defecto del DAG, como reintentos (retries) y notificaciones por correo (email_on_failure).

Falar no WhatsApp