Automatiza Pipelines en GCP con Cloud Composer y DAGs

Recapitulación: ¿Qué es un DAG?

Un DAG (Directed Acyclic Graph) es un script de Python que define un flujo de trabajo en Airflow (el motor de Cloud Composer). La "magia" de los DAGs reside en su capacidad para definir tareas y las dependencias entre ellas, asegurando que se ejecuten en el orden correcto. Un DAG debe cumplir dos condiciones:

• Dirigido (Directed): Las dependencias tienen un sentido claro. La Tarea A va antes que la Tarea B.
• Acíclico (Acyclic): No puede haber bucles o dependencias circulares (ej. A depende de B, y B depende de A).

Los Bloques de Construcción: Operadores

Dentro de un DAG, cada tarea es instanciada por un Operador. Airflow viene con una enorme biblioteca de operadores listos para usar que saben cómo interactuar con diferentes sistemas. Para un Ingeniero de Datos en GCP, los más importantes son:

• `DataflowCreatePythonJobOperator` o `DataflowTemplatedJobStartOperator`: Para iniciar un pipeline de Dataflow (ya sea desde un script de Python o una plantilla).
• `BigQueryInsertJobOperator` o `BigQueryCheckOperator`: Para ejecutar consultas de BigQuery, realizar comprobaciones de calidad de datos, o ejecutar procedimientos almacenados.
• `BashOperator`: Para ejecutar cualquier comando de shell. Es increíblemente versátil para tareas sencillas.
• `PythonOperator`: Para ejecutar una función de Python.

Construyendo un DAG Realista

Vamos a diseñar un DAG que represente un flujo de trabajo de ELT común:

1. Tarea 1: Iniciar un pipeline de Dataflow (`DataflowOperator`). Este pipeline lee datos de una fuente externa y los carga en una tabla "raw" en BigQuery.
2. Tarea 2: Ejecutar una transformación SQL (`BigQueryInsertJobOperator`). Esta tarea se ejecuta después de que el pipeline de Dataflow termine con éxito. Lee de la tabla "raw", aplica lógica de negocio (limpieza, enriquecimiento) y materializa el resultado en una tabla de "análisis".
3. Tarea 3: Realizar una comprobación de calidad (`BigQueryCheckOperator`). Después de la transformación, esta tarea verifica que los datos en la tabla de análisis cumplen ciertas condiciones (ej. no hay nulos en columnas clave, no hay valores duplicados).
4. Tarea 4: Tarea final. Si todas las comprobaciones son correctas, el pipeline finaliza con éxito. Si alguna falla, la tarea se marca como fallida, y Airflow puede enviar una alerta.

En nuestro script DAG de Python, definiremos estas cuatro tareas y estableceremos las dependencias de la siguiente manera: Tarea 1 >> Tarea 2 >> Tarea 3 >> Tarea 4. La sintaxis de Airflow (`>>` y `<<`) hace que definir estas cadenas de dependencias sea muy intuitivo.

Manejando Fallos y Reintentos

Una de las mayores ventajas de un orquestador es su capacidad para gestionar fallos. En la definición de un DAG, podemos configurar parámetros como:

• `retries`: El número de veces que Airflow intentará re-ejecutar una tarea si falla.
• `retry_delay`: El tiempo que esperará entre reintentos.
• `on_failure_callback`: Una función que se ejecutará si una tarea falla definitivamente (ej. para enviar una alerta a Slack o por correo electrónico).

Conclusión: El Director de la Orquesta de Datos

Cloud Composer y Apache Airflow transforman una colección de scripts y jobs aislados en un sistema de datos coherente, robusto y automatizado. Dominar la creación de DAGs, el uso de operadores y la configuración de dependencias es lo que te permite pasar de simplemente "ejecutar código" a "construir y gestionar arquitecturas de datos de producción". Es la habilidad que garantiza que tus pipelines se ejecuten de manera fiable, predecible y mantenible día tras día.