通过 Airflow 的 Postgres 不返回查询结果 - 冻结直到超时

Question

我在 docker 上使用气流 2.2.0 并通过 ETL 管道发出多个查询。一切正常，直到我达到如下所示的特定查询：

INSERT INTO final_table (
    col1,
    col2
)
SELECT
    col1,
    col2
FROM some_temp_table
LEFT JOIN some_other_tables

在我拥有的较大的表集上完成此查询大约需要 30 秒。但是，发生的情况是查询是由 Airflow 向 Postgres 发出的，然后在大约 5 分钟内什么都没有返回，直到 Airflow 最终由于我相信在该特定任务中没有收到任何响应而导致管道崩溃。

错误如下：

psycopg2.OperationalError: server closed the connection unexpectedly
    This probably means the server terminated abnormally
    before or while processing the request.
    (...)
    self._execute_queries(connection=connection, cursor=cursor, query=query)
psycopg2.InterfaceError: connection already closed

我已经检查了所有我能想到的东西，但没有发现任何奇怪的东西。 pg_stat_activity 表显示查询始终处于活动状态，似乎没有被任何其他查询阻塞。

Postgres 的日志显示，Airflow 发出的查询被 Postgres 接收。日志如下：

[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

但在那之后，postgres 不会发出任何 COMMIT 命令（因为它发生在所有其他事务中）。查询似乎无限期地挂在那个状态。

当我直接通过数据库客户端（在本例中为 DBeaver）运行相同的精确查询时，一切正常，查询开始运行，表在 30 秒内填满。此外，当我使用较小的数据集运行相同的 ETL 管道时，一切正常，查询通过 Airflow 成功运行。

我真的很困惑问题可能是什么以及我应该将注意力集中在哪里。如果有人经历过类似的事情，我们将不胜感激。

Answer 1

您使用的是 PostgresOperator 还是 PostgresHook？

您是否启用了自动提交标志？

看看这个例子，看看它是否有帮助。

import datetime

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

# create_pet_table, populate_pet_table, get_all_pets, and get_birth_date are examples of tasks created by
# instantiating the Postgres Operator

with DAG(
    dag_id="postgres_operator_dag",
    start_date=datetime.datetime(2020, 2, 2),
    schedule_interval="@once",
    catchup=False,
) as dag:
    # [START postgres_operator_howto_guide_create_pet_table]
    create_pet_table = PostgresOperator(
        task_id="create_pet_table",
        sql="""
            CREATE TABLE IF NOT EXISTS pet (
            pet_id SERIAL PRIMARY KEY,
            name VARCHAR NOT NULL,
            pet_type VARCHAR NOT NULL,
            birth_date DATE NOT NULL,
            OWNER VARCHAR NOT NULL);
          """,
    )
    # [END postgres_operator_howto_guide_create_pet_table]
    # [START postgres_operator_howto_guide_populate_pet_table]
    populate_pet_table = PostgresOperator(
        task_id="populate_pet_table",
        sql="""
            INSERT INTO pet (name, pet_type, birth_date, OWNER)
            VALUES ( 'Max', 'Dog', '2018-07-05', 'Jane');
            INSERT INTO pet (name, pet_type, birth_date, OWNER)
            VALUES ( 'Susie', 'Cat', '2019-05-01', 'Phil');
            INSERT INTO pet (name, pet_type, birth_date, OWNER)
            VALUES ( 'Lester', 'Hamster', '2020-06-23', 'Lily');
            INSERT INTO pet (name, pet_type, birth_date, OWNER)
            VALUES ( 'Quincy', 'Parrot', '2013-08-11', 'Anne');
            """,
    )
    # [END postgres_operator_howto_guide_populate_pet_table]
    # [START postgres_operator_howto_guide_get_all_pets]
    get_all_pets = PostgresOperator(task_id="get_all_pets", sql="SELECT * FROM pet;")
    # [END postgres_operator_howto_guide_get_all_pets]
    # [START postgres_operator_howto_guide_get_birth_date]
    get_birth_date = PostgresOperator(
        task_id="get_birth_date",
        sql="""
            SELECT * FROM pet
            WHERE birth_date
            BETWEEN SYMMETRIC DATE '{{ params.begin_date }}' AND DATE '{{ params.end_date }}';
            """,
        params={'begin_date': '2020-01-01', 'end_date': '2020-12-31'},
    )
    # [START postgres_operator_howto_guide_get_birth_date]

    create_pet_table >> populate_pet_table >> get_all_pets >> get_birth_date
    # [END postgres_operator_howto_guide]

教程：https://registry.astronomer.io/dags/example-postgres