Quiz: Operadores, Sensores y su Aplicación en ETL

Concepto clave

En Apache Airflow, los operadores son las unidades básicas de trabajo que ejecutan tareas específicas dentro de un DAG. Piensa en ellos como los trabajadores especializados en una línea de producción: cada uno sabe exactamente qué hacer (extraer datos, transformarlos, cargarlos) y solo necesita las instrucciones correctas para ejecutarse.

Los sensores, por otro lado, son operadores especializados que esperan por una condición externa antes de continuar. Imagina un inspector de calidad que verifica que la materia prima llegó antes de permitir que comience el proceso de transformación. En ETL, esto se traduce en esperar por la llegada de un archivo, la disponibilidad de una base de datos, o la finalización de otro proceso.

Cómo funciona en la práctica

Veamos un ejemplo paso a paso de un pipeline ETL que procesa datos de ventas diarias:

1. Un FileSensor monitorea un directorio FTP esperando el archivo ventas_diarias.csv
2. Cuando el archivo aparece, un PythonOperator extrae los datos y los valida
3. Un BashOperator ejecuta un script de transformación que calcula métricas clave
4. Finalmente, un PostgresOperator carga los resultados transformados a la base de datos de reporting

Cada operador se define con sus parámetros específicos y se encadena mediante dependencias para crear el flujo completo.

Caso de estudio

Una empresa de retail necesita procesar datos de inventario desde múltiples fuentes:

El sensor S3KeySensor redujo los errores por archivos incompletos en un 85% al verificar que el archivo tenga el tamaño esperado antes de procesarlo.

Errores comunes

• Timeout demasiado corto en sensores: Configurar timeout muy bajo hace que los sensores fallen prematuramente. Solución: Calcular el tiempo máximo realista de espera y agregar un margen del 20%.
• Operadores bloqueantes en bucles: Usar while loops dentro de PythonOperator puede bloquear el scheduler. Solución: Implementar lógica asíncrona o usar sensores especializados.
• Falta de manejo de fallos: No configurar retries y retry_delay en operadores críticos. Solución: Definir políticas de reintento basadas en la criticidad de cada tarea.
• Dependencias circulares: Crear ciclos donde el operador A depende de B y B depende de A. Solución: Usar herramientas de validación de DAGs antes del despliegue.

Checklist de dominio

1. Puedo seleccionar el operador correcto para cada tipo de tarea ETL (extracción, transformación, carga)
2. Sé configurar sensores con condiciones específicas y tiempos de espera apropiados
3. Entiendo cómo encadenar operadores y sensores para crear flujos lógicos
4. Puedo implementar manejo de errores y políticas de reintento
5. Conozco la diferencia entre operadores estándar y custom operators
6. Sé monitorear y debuggear la ejecución de operadores en tiempo real
7. Puedo optimizar el uso de recursos entre operadores concurrentes

Implementa un pipeline ETL con sensores y operadores especializados

Construye un DAG que procese datos meteorológicos desde múltiples fuentes:

1. Crea un sensor que espere por un archivo JSON en /data/weather con el patrón weather_*.json
2. Implementa un operador que valide la estructura del JSON (debe contener temperatura, humedad y timestamp)
3. Agrega un operador de transformación que convierta temperaturas de Fahrenheit a Celsius
4. Configura un operador que cargue los datos validados y transformados a una tabla PostgreSQL
5. Define dependencias apropiadas entre todas las tareas
6. Agrega manejo de errores con 3 reintentos y delay exponencial

Entrega el código Python completo del DAG.

• Usa FileSensor con mode='poke' para el monitoreo de archivos
• Considera usar JsonSchemaValidatorOperator para validación estructural
• Recuerda que PostgreSQLOperator requiere configuración de conexión previa

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.