Implementar Branching y Condicionales en DAGs

Tipo de lección: Video

Concepto clave

En Apache Airflow, el branching y los condicionales permiten crear flujos de trabajo dinámicos donde las tareas no siempre se ejecutan en secuencia lineal. Imagina que estás construyendo una autopista con múltiples salidas: dependiendo del tráfico (condición), algunos carriles se dirigen a una ciudad, otros a otra, pero todos convergen eventualmente. En Airflow, esto significa que tu DAG puede tomar decisiones en tiempo de ejecución sobre qué rama de tareas ejecutar, basándose en datos, resultados previos o parámetros externos.

Esto es crucial para pipelines ETL complejos porque los datos rara vez son uniformes. Por ejemplo, al procesar datos de ventas, podrías necesitar rutas diferentes para transacciones normales versus devoluciones, o para datos de diferentes regiones. Los condicionales te permiten modelar esta lógica directamente en el DAG, haciendo que tu orquestación sea más inteligente y adaptable.

El branching no es solo sobre elegir una rama; es sobre crear pipelines que se adapten a la realidad de tus datos.

Cómo funciona en la práctica

Airflow implementa branching principalmente a través del operador BranchPythonOperator, que evalúa una función Python para decidir qué tarea(s) ejecutar a continuación. La función debe devolver el task_id de la siguiente tarea o una lista de task_ids. Aquí un ejemplo paso a paso:

1. Define una función Python que tome el contexto de Airflow (por ejemplo, usando **kwargs) y devuelva un task_id basado en una condición.
2. Crea un BranchPythonOperator que use esa función.
3. Configura las dependencias: el branching operator debe estar aguas arriba de las tareas que representan las ramas posibles.
4. Las tareas no seleccionadas se marcan como skipped automáticamente.

Ejemplo básico con datos:

from airflow.operators.python import BranchPythonOperator

def decide_ruta(**kwargs):
    ti = kwargs['ti']
    # Supongamos que una tarea anterior pusó un valor en XCom
    valor = ti.xcom_pull(task_ids='tarea_anterior', key='clave')
    if valor > 100:
        return 'procesar_grande'
    else:
        return 'procesar_pequeno'

branch_task = BranchPythonOperator(
    task_id='decidir_ruta',
    python_callable=decide_ruta,
    dag=dag
)

Caso de estudio

Imagina un pipeline ETL para procesar datos de sensores IoT en una fábrica. Los sensores envían lecturas de temperatura cada minuto. Tu objetivo es: si la temperatura supera un umbral crítico (ej., 80°C), activar una alerta inmediata y registrar en una base de datos de incidentes; si está en rango normal, simplemente agregar los datos para reportes diarios.

Estructura del DAG:

• extraer_datos: Lee los datos del sensor desde un API.
• branch_temperatura: Usa BranchPythonOperator para evaluar si la temperatura > 80°C.
• Rama A: activar_alerta y registrar_incidente (si temperatura alta).
• Rama B: agregar_datos (si temperatura normal).
• generar_reporte: Tarea común que se ejecuta después de cualquiera de las ramas, usando trigger_rule='none_failed' para asegurar ejecución.

Tabla de umbrales de ejemplo:

Temperatura (°C)	Acción	Tarea Asignada
> 80	Alerta crítica	activar_alerta
50-80	Advertencia	registrar_incidente
< 50	Normal	agregar_datos

Este caso muestra cómo el branching maneja escenarios del mundo real donde los datos dictan el flujo de trabajo.

Errores comunes

• Olvidar configurar trigger_rule en tareas aguas abajo: Si una rama se salta, las tareas comunes podrían no ejecutarse. Usa trigger_rule='none_failed' o 'dummy' para asegurar continuidad.
• Devolver múltiples task_ids sin entender las consecuencias: El BranchPythonOperator puede devolver una lista, pero esto ejecuta todas esas tareas en paralelo, no como ramas exclusivas. Asegúrate de que tu lógica coincida con la intención.
• No manejar casos edge en la función de branching: Siempre incluye un caso por defecto o validación para evitar errores en tiempo de ejecución.
• Confundir branching con condicionales en tareas individuales: El branching decide entre tareas; para lógica dentro de una tarea, usa condicionales Python normales en el operador.
• Ignorar el estado skipped: Las tareas no seleccionadas se marcan como skipped, lo que puede afectar monitoreo. Planifica cómo manejar estos en tus dashboards.

Checklist de dominio

1. Puedo explicar la diferencia entre BranchPythonOperator y un operador condicional dentro de una tarea.
2. He implementado un DAG con al menos dos ramas basadas en datos dinámicos (ej., desde XCom).
3. Sé cómo configurar trigger_rule para tareas que dependen de ramas skipped.
4. Puedo debuggear un DAG donde el branching no selecciona la rama esperada.
5. He usado branching para manejar al menos un escenario real como errores de datos o umbrales.
6. Entiendo cómo las tareas skipped afectan el flujo general y el reporting.
7. Puedo documentar la lógica de branching para que otros ingenieros la entiendan.

Construye un DAG con Branching para Procesamiento de Pedidos E-commerce

En este ejercicio, crearás un DAG en Apache Airflow que simule un pipeline ETL para procesar pedidos en una plataforma de e-commerce. El DAG debe usar branching para manejar diferentes tipos de pedidos basados en su valor total.

1. Configura el entorno: Asegúrate de tener Airflow instalado y funcionando. Crea un nuevo archivo Python para tu DAG en la carpeta dags/.
2. Define las tareas: Crea las siguientes tareas usando operadores de Python:
   • extraer_pedidos: Simula extraer datos de pedidos (puedes usar una lista de diccionarios con order_id, total_amount).
   • branch_pedido: Usa BranchPythonOperator para decidir la ruta basada en total_amount: si > 500, ir a procesar_grande; si <= 500, ir a procesar_pequeno.
   • procesar_grande: Simula procesamiento especial para pedidos grandes (ej., log a consola).
   • procesar_pequeno: Simula procesamiento estándar para pedidos pequeños.
   • notificar_fin: Tarea común que se ejecute después de cualquier rama, notificando finalización.
3. Implementa la lógica de branching: En la función para branch_pedido, usa XCom para obtener total_amount desde extraer_pedidos y devuelve el task_id correspondiente.
4. Configura dependencias y trigger rules: Establece las dependencias entre tareas y asegúrate de que notificar_fin use trigger_rule='none_failed'.
5. Prueba el DAG: Ejecuta el DAG en Airflow y verifica que las ramas se seleccionen correctamente basadas en los datos. Prueba con al menos dos pedidos: uno con total 600 y otro con total 300.
6. Documenta: Agrega comentarios en el código explicando tu lógica de branching.

Pistas

• Usa ti.xcom_pull en la función de branching para acceder a datos de tareas anteriores.
• Recuerda que BranchPythonOperator devuelve un string con el task_id, no el objeto de la tarea.
• Si notificar_fin no se ejecuta, revisa la configuración de trigger_rule.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.