Quiz: Integración y Automatización ETL

Concepto clave

La integración de Apache Airflow con herramientas de datos y la automatización ETL se centra en orquestar flujos de trabajo complejos donde múltiples sistemas deben colaborar de manera coordinada. Imagina que eres el director de una orquesta: Airflow es el director que marca el tempo y coordina a cada músico (herramienta), asegurando que todos toquen en el momento correcto y en armonía. En este contexto, las herramientas de datos pueden ser bases de datos, servicios en la nube, APIs o plataformas de procesamiento como Spark.

La automatización ETL va más allá de simplemente ejecutar tareas; implica gestionar dependencias, manejar errores, y garantizar que los datos fluyan de manera confiable desde las fuentes hasta los destinos. Un pipeline bien integrado se adapta a cambios en los datos o en las infraestructuras, manteniendo la consistencia y calidad. La clave está en diseñar DAGs que no solo ejecuten código, sino que también monitoreen y respondan al estado de las herramientas externas.

Cómo funciona en la práctica

Para integrar Airflow con una herramienta como Amazon Redshift, sigue estos pasos:

1. Configura un conector (connection) en Airflow para Redshift, almacenando credenciales de forma segura en la interfaz web o mediante variables.
2. Crea un operador personalizado o usa uno existente, como PostgresOperator adaptado para Redshift, dentro de un DAG.
3. Define tareas que ejecuten consultas SQL en Redshift, por ejemplo, para cargar datos desde S3 o transformar tablas.
4. Establece dependencias entre tareas usando >> para asegurar que los datos se procesen en el orden correcto.
5. Configura alertas y reintentos para manejar fallos de conexión o errores en las consultas.

Ejemplo de una tarea simple en un DAG:

from airflow.providers.postgres.operators.postgres import PostgresOperator

load_data = PostgresOperator(
    task_id='load_to_redshift',
    postgres_conn_id='redshift_conn',
    sql='COPY sales FROM ''s3://bucket/data.csv'' CREDENTIALS ''aws_iam_role=arn:aws:iam::123456789012:role/RedshiftRole'' CSV;',
    dag=dag
)

Caso de estudio

Una empresa de e-commerce necesita automatizar su pipeline de análisis de ventas diarias. Los datos provienen de múltiples fuentes: una base de datos MySQL con transacciones, un bucket S3 con logs de clics, y una API externa de clima. El objetivo es consolidar estos datos en un data warehouse de Google BigQuery para reportes.

Implementación con Airflow:

• DAG programado para ejecutarse cada día a las 2 AM.
• Tarea 1: Extraer datos de MySQL usando MySqlOperator y guardarlos en un archivo Parquet en S3.
• Tarea 2: Descargar logs de S3 y transformarlos con un script Python en un PythonOperator.
• Tarea 3: Llamar a la API de clima con SimpleHttpOperator y parsear la respuesta JSON.
• Tarea 4: Cargar todos los datos transformados a BigQuery usando BigQueryOperator.
• Tarea 5: Enviar un correo con EmailOperator si el pipeline falla.

Este caso reduce el tiempo de procesamiento manual de 4 horas a 30 minutos automatizado, con monitoreo en tiempo real.

Errores comunes

• No gestionar conexiones de forma segura: Almacenar credenciales en código plano puede comprometer la seguridad. Usa los conectores de Airflow o herramientas como AWS Secrets Manager.
• Sobrecargar un solo DAG con demasiadas tareas: Esto dificulta el debugging y el mantenimiento. Divide pipelines complejos en DAGs más pequeños y modulares.
• Ignorar el manejo de errores: Sin reintentos o alertas, los fallos en herramientas externas pueden detener todo el pipeline. Configura retries y on_failure_callback en los operadores.
• No probar integraciones en un entorno de staging: Ejecutar DAGs directamente en producción puede causar pérdida de datos. Valida con datos de prueba primero.
• Olvidar limpiar recursos temporales: Archivos o conexiones no cerradas pueden llevar a fugas de memoria. Usa context managers o tareas de limpieza al final del DAG.

Checklist de dominio

1. ¿Puedes configurar un conector en Airflow para una herramienta como Redshift o BigQuery?
2. ¿Sabes diseñar un DAG que orqueste al menos tres herramientas diferentes (ej., base de datos, API, almacenamiento en la nube)?
3. ¿Implementas manejo de errores con reintentos y notificaciones en tus pipelines?
4. ¿Eres capaz de optimizar el rendimiento usando pools y concurrencia en Airflow?
5. ¿Puedes integrar Airflow con un sistema de monitoreo como Prometheus o Datadog?
6. ¿Sabes escribir operadores personalizados para herramientas no soportadas nativamente?
7. ¿Documentas tus DAGs y flujos de trabajo para facilitar el mantenimiento en equipo?

Crear un DAG que integre MySQL, una API REST y S3

En este ejercicio, desarrollarás un DAG en Apache Airflow que automatice un pipeline ETL simple, integrando múltiples herramientas comunes en entornos de datos.

1. Configura el entorno: Asegúrate de tener Airflow instalado y corriendo localmente o en un entorno de prueba. Crea conexiones en la interfaz web de Airflow para MySQL (usando un conector como mysql_default) y para AWS S3 (usando aws_default). Para la API, puedes simularla con un endpoint público como JSONPlaceholder.
2. Diseña el DAG: Crea un archivo Python para el DAG, definiendo un ID único y una programación diaria. El DAG debe incluir tres tareas principales:
   • Tarea 1: Extraer datos de una tabla en MySQL (ej., SELECT * FROM orders WHERE date = '{{ ds }}') y guardar el resultado en un archivo CSV temporal.
   • Tarea 2: Llamar a una API REST (ej., https://jsonplaceholder.typicode.com/posts/1) para obtener datos adicionales, como información de usuario, y combinarlos con los datos de MySQL en un DataFrame de Pandas.
   • Tarea 3: Subir el archivo combinado a un bucket de S3 en formato Parquet, usando la biblioteca boto3 o un operador de Airflow.
3. Implementa las tareas: Usa operadores apropiados:
   • Para MySQL, emplea MySqlOperator o PythonOperator con una conexión.
   • Para la API, usa SimpleHttpOperator o PythonOperator con requests.
   • Para S3, utiliza PythonOperator con boto3 o S3Hook de Airflow.
   Asegúrate de definir dependencias entre tareas (ej., Tarea 1 >> Tarea 2 >> Tarea 3).
4. Añade robustez: Configura al menos un reintento para cada tarea en caso de fallos de red o tiempo de espera. Agrega un manejador de errores que registre incidentes en un archivo de log.
5. Prueba y valida: Ejecuta el DAG en Airflow y verifica que los datos se procesen correctamente. Revisa los logs para asegurarte de que no haya errores y que el archivo se suba a S3.

• Usa variables de Airflow para almacenar URLs o nombres de buckets de manera dinámica y segura.
• Para la combinación de datos en la Tarea 2, considera usar Pandas para manipular DataFrames fácilmente.
• Si encuentras problemas de permisos con S3, verifica que la conexión de AWS tenga las políticas IAM adecuadas.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.