Ejercicio: Construir un DAG con Múltiples Ramas y Fallback

Concepto clave

En pipelines ETL complejos, los flujos de trabajo rara vez son lineales. A menudo necesitas ramificaciones para manejar diferentes escenarios de datos y mecanismos de fallback para garantizar la resiliencia del sistema. Imagina una cadena de montaje en una fabrica: cuando un producto llega a un punto de decision, puede seguir diferentes lineas segun su tipo, pero siempre hay una ruta alternativa si la principal falla.

En Apache Airflow, esto se implementa mediante operadores como BranchPythonOperator que decide que camino seguir, combinado con trigger rules que controlan como se ejecutan las tareas dependiendo del estado de sus predecesoras. El fallback es como tener un plan B automatizado: si una tarea critica falla, el sistema automaticamente ejecuta una alternativa sin detener todo el pipeline.

Como funciona en la practica

Vamos a construir un ejemplo paso a paso. Supongamos que procesamos datos de ventas que pueden venir en tres formatos: CSV, JSON o XML. Nuestro DAG debe:

1. Detectar el formato del archivo de entrada
2. Ramificar a la ruta de procesamiento correspondiente
3. Tener un procesamiento de fallback si ningun formato es reconocido
4. Unificar los resultados

Primero, definimos la funcion de ramificacion:

def decide_ruta(**context):
    file_path = context['ti'].xcom_pull(task_ids='detectar_formato')
    if file_path.endswith('.csv'):
        return 'procesar_csv'
    elif file_path.endswith('.json'):
        return 'procesar_json'
    elif file_path.endswith('.xml'):
        return 'procesar_xml'
    else:
        return 'formato_desconocido'

Luego configuramos las trigger rules. Para la tarea de fallback, usamos trigger_rule='all_failed' para que solo se ejecute si todas las ramas principales fallan.

Caso de estudio

Una empresa de e-commerce procesa diariamente 1TB de datos de clics. El pipeline debe:

El DAG implementado tiene esta estructura:

detectar_calidad -> [calidad_alta, calidad_media, calidad_baja]
calidad_alta -> procesamiento_completo
calidad_media -> procesamiento_basico -> alertar_equipo
calidad_baja -> usar_datos_anteriores -> generar_reporte
[procesamiento_completo, procesamiento_basico, usar_datos_anteriores] -> consolidar_resultados

La tarea alertar_equipo tiene trigger_rule='one_success' para ejecutarse si calidad_media tiene exito, mientras que generar_reporte tiene trigger_rule='all_done' para ejecutarse siempre.

Errores comunes

• Olvidar trigger rules personalizadas: Por defecto, las tareas esperan que todas sus predecesoras tengan exito. Si no configuras trigger_rule='none_failed' o similar en las rutas de consolidacion, el DAG se quedara atascado cuando una rama falle.
• Dependencias circulares: Crear ramas que eventualmente dependen de si mismas, causando loops infinitos. Siempre valida que cada camino tenga un final claro.
• Manejo inadecuado de XCom: Las ramas pueden necesitar compartir datos. Usa XCom con keys especificas para evitar contaminacion cruzada entre rutas paralelas.
• Falta de monitoreo de ramas inactivas: Si una rama nunca se ejecuta porque la condicion nunca se cumple, podrias perder datos. Implementa logging y metricas para cada posible camino.
• Fallback que falla: El mecanismo de respaldo debe ser extremadamente simple y confiable. Si tu fallback es tan complejo como la ruta principal, estas duplicando puntos de falla.

Checklist de dominio

1. Puedo diseñar un DAG con al menos tres ramas condicionales usando BranchPythonOperator
2. Se configurar trigger rules diferentes para tareas de consolidacion, alertas y fallback
3. Implemento mecanismos de fallback que se activan automaticamente cuando las rutas principales fallan
4. Uso XCom de forma segura para compartir datos entre ramas paralelas
5. Incluyo logging diferenciado para cada rama del DAG
6. Valido que no existan dependencias circulares en el grafo de tareas
7. Documento claramente todas las posibles rutas de ejecucion y sus condiciones

Construye un DAG para procesamiento de datos meteorologicos con ramificaciones y fallback

Objetivo: Crear un DAG que procese datos meteorologicos de diferentes fuentes con mecanismos de resiliencia.

Contexto: Tu empresa recibe datos de tres APIs meteorologicas diferentes. Cada dia debes:

1. Intentar obtener datos de la API primaria (OpenWeatherMap)
2. Si falla, intentar con la API secundaria (WeatherAPI)
3. Si ambas fallan, usar datos historicos locales como fallback
4. Procesar los datos obtenidos segun su formato (JSON o XML)
5. Almacenar resultados en una base de datos

Pasos:

1. Crea un DAG llamado 'procesamiento_meteorologico' con schedule_interval diario
2. Implementa una tarea inicial 'obtener_datos_api_primaria' que simule llamar a una API (puedes usar PythonOperator con una funcion que a veces falle aleatoriamente)
3. Crea un BranchPythonOperator 'decidir_fuente_datos' que:
   • Si la API primaria tuvo exito: continuar con 'procesar_datos_primarios'
   • Si fallo: ejecutar 'obtener_datos_api_secundaria'
   • Si ambas APIs fallaron: ejecutar 'usar_datos_historicos'
4. Configura las trigger rules para que 'procesar_datos' se ejecute cuando cualquier fuente tenga exito (trigger_rule='one_success')
5. Agrega una rama adicional: si los datos son JSON, procesarlos con transformaciones complejas; si son XML, procesarlos con transformaciones basicas
6. Implementa alertas: si se usan datos historicos, enviar una notificacion (simulada) al equipo
7. Prueba tu DAG con diferentes escenarios de fallo

• Usa un diccionario en XCom para pasar el estado de cada API y el formato de datos entre tareas
• Para simular fallos aleatorios, usa random.randint(0,1) en tus funciones y maneja las excepciones apropiadamente
• Considera usar ShortCircuitOperator como alternativa a BranchPythonOperator para algunas decisiones simples

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.