Integración de Apache Airflow con AWS SQS y Oracle Database

Cuando necesitas orquestar flujos de datos entre sistemas cloud y bases de datos on-premise, la combinación de Apache Airflow, AWS SQS y Oracle Database ofrece una arquitectura robusta y desacoplada. Este patrón permite procesar mensajes de forma confiable, con reintentos automáticos y visibilidad completa del pipeline.

Arquitectura y decisión: Airflow vs AWS Lambda

El flujo típico comienza con un sensor de Airflow (SqsSensor) que detecta mensajes en la cola SQS. Al recibirlos, se dispara un DAG que extrae el payload, lo transforma y lo escribe en tablas Oracle. La configuración del sensor con mode='wait_for_callbacks' permite polling eficiente sin saturar la cola, mientras que la conexión AWS se define mediante AWS_CONN_ID con políticas IAM mínimas (sqs:ReceiveMessage, sqs:DeleteMessage).

Aquí surge una decisión arquitectónica clave: ¿cuándo usar Apache Airflow en lugar de AWS Lambda? Airflow es superior cuando se requiere orquestación compleja con múltiples pasos, reintentos automáticos configurables, transaccionalidad explícita y visibilidad del flujo de trabajo completo. Por ejemplo, si cada mensaje de SQS debe validarse, enriquecerse con datos de otra API y luego insertarse en Oracle con control de transacciones, Airflow maneja las dependencias y fallos de forma nativa. AWS Lambda es adecuado para eventos simples y sin estado, donde la latencia debe ser mínima y la carga esporádica. Para procesamiento batch controlado con Oracle como destino, Airflow ofrece un control granular que Lambda no puede igualar.

Configuración del operador Oracle para escritura transaccional

La escritura en Oracle Database se realiza mediante OracleOperator o OracleHook, utilizando el driver cx_Oracle. La conexión requiere configurar oracle_home y tns_admin para apuntar al archivo tnsnames.ora, lo que permite conectarse a instancias RAC o standalone. Es crítico manejar las transacciones explícitamente: tras procesar un lote de mensajes, se ejecuta commit para evitar duplicados en caso de fallo del DAG. La estrategia de reintentos se define con retries=3 y retry_delay=timedelta(seconds=30), tolerando bloqueos temporales en Oracle o problemas de red.

Control de carga y desacoplamiento con SQS

Para evitar sobrecargar Oracle, se configura max_messages=10 en el SqsSensor, limitando el tamaño del lote por ejecución del DAG. El VisibilityTimeout de SQS (por ejemplo, 600 segundos) evita que los mensajes sean reprocesados mientras el DAG escribe en la base de datos. Si un mensaje falla tras max_receive_count=5, se envía a una Dead Letter Queue (DLQ), donde un DAG separado puede auditar las causas del fallo. Este patrón de backpressure asegura que el sistema se adapte a picos de carga sin colapsar Oracle.

Monitoreo y validación de integridad

Los logs estructurados de Airflow se exportan a CloudWatch Logs, permitiendo correlacionar IDs de mensaje SQS con transacciones Oracle. Se implementan métricas personalizadas con StatsD o AirflowMetrics para contar mensajes procesados vs. fallidos, con alertas si la cola SQS supera un umbral. Un DAG de reconciliación diaria compara el atributo ApproximateNumberOfMessages de SQS con los registros insertados en Oracle, validando la integridad del pipeline.