Postgres a través de Airflow no está devolviendo el resultado de la consulta – queda bloqueado hasta el tiempo de espera.
Estoy utilizando Airflow 2.2.0 en Docker y ejecutando varias consultas a través de un pipeline ETL. Todo funciona bien hasta que llego a una consulta en particular que se ve así:
INSERT INTO final_table (
col1,
col2
)
SELECT
col1,
col2
FROM some_temp_table
LEFT JOIN some_other_tables
Esta consulta debería tardar alrededor de 30 segundos en completarse en el conjunto de tablas más grande que tengo. Sin embargo, lo que sucede es que Airflow emite la consulta a Postgres y luego no recibe ninguna respuesta durante aproximadamente 5 minutos hasta que Airflow finalmente interrumpe el pipeline debido a que no recibe ninguna respuesta en esa tarea en particular, creo.
El error es el siguiente:
psycopg2.OperationalError: el servidor cerró la conexión inesperadamente
Esto probablemente significa que el servidor terminó anormalmente
antes o durante el procesamiento de la solicitud.
(...)
self._execute_queries(connection=connection, cursor=cursor, query=query)
psycopg2.InterfaceError: conexión ya cerrada
He revisado todo lo que se me ocurrió y no puedo detectar nada extraño. La tabla pg_stat_activity muestra que la consulta siempre está en estado activo y no parece estar bloqueada por ninguna otra consulta.
Los registros de Postgres muestran que la consulta emitida por Airflow es recibida por Postgres. Los registros son los siguientes:
[3244] LOG: statement: BEGIN
[3244] LOG: statement:
INSERT INTO final_table (
col1,
col2
)
SELECT
col1,
col2
FROM some_temp_table
LEFT JOIN some_other_tables
Pero después de eso, no se emite ningún comando COMMIT por parte de Postgres (como ocurre con todas las demás transacciones). La consulta parece quedarse en ese estado indefinidamente.
Cuando ejecuto esta misma consulta directamente a través del cliente de la base de datos (en este caso, DBeaver), todo funciona bien y la consulta se ejecuta y la tabla se llena en 30 segundos. Además, cuando ejecuto el mismo pipeline ETL pero con un conjunto de datos más pequeño, todo funciona bien y la consulta se ejecuta correctamente a través de Airflow.
Realmente estoy confundido acerca de cuál puede ser el problema y en qué debo centrar mi atención. Si alguien ha pasado por algo similar, cualquier ayuda sería muy apreciada.
davy.ai
Parece que el problema aquí está con Airflow en lugar de Postgres. Dado que la consulta funciona correctamente cuando se ejecuta directamente a través del cliente de base de datos, podría haber un problema con la forma en que Airflow está manejando la consulta o la conexión a Postgres.
Una cosa para verificar serían los registros de trabajador y programador de Airflow para ver si hay errores o advertencias relacionadas con la tarea en particular que está fallando. Podría haber un tiempo de espera u otro problema con la forma en que Airflow está gestionando la tarea.
Otra cosa para verificar sería el archivo de configuración de Airflow para asegurarse de que las configuraciones para la conexión a Postgres estén correctamente configuradas. Podría valer la pena comparar las configuraciones con las utilizadas en el cliente de base de datos para ver si hay alguna diferencia.
Además, podría valer la pena experimentar con diferentes configuraciones para la tarea de Airflow, como ajustar el tiempo de espera o cambiar la forma en que Airflow gestiona las conexiones a Postgres.
En general, parece que el problema probablemente esté relacionado con la interacción entre Airflow y Postgres, en lugar de un problema con la consulta en sí o la base de datos. Al analizar cuidadosamente los registros y la configuración, debería ser posible identificar la fuente del problema y encontrar una solución.