能否仅通过PostgresOperator在Airflow任务日志中展示PostgreSQL终端日志？

1. 调整SQL脚本主动输出日志 #

在SQL开头设置客户端消息级别为notice，确保PostgreSQL将NOTICE消息发送到客户端；同时通过GET DIAGNOSTICS获取受影响行数，用RAISE NOTICE输出：

SET client_min_messages = notice;

-- 业务SQL示例
UPDATE your_table 
SET status = 'processed' 
WHERE created_at < NOW() - INTERVAL '7 days';

-- 获取并输出受影响行数
GET DIAGNOSTICS updated_rows = ROW_COUNT;
RAISE NOTICE '已更新 % 条记录', updated_rows;

RAISE NOTICE '任务执行完成';

PL/pgSQL块写法：

SET client_min_messages = notice;

DO $$
DECLARE
    deleted_rows integer;
BEGIN
    DELETE FROM expired_data WHERE expiry_date < NOW();
    GET DIAGNOSTICS deleted_rows = ROW_COUNT;
    RAISE NOTICE '已删除 % 条过期数据', deleted_rows;
END $$;

2. 自定义PostgresOperator捕获NOTICE #

默认PostgresOperator不会把PostgreSQL的NOTICE写入Airflow日志，继承原Operator添加日志捕获逻辑，无需改用PythonOperator：

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

class LoggingPostgresOperator(PostgresOperator):
    def execute(self, context):
        hook = self.get_hook()
        conn = hook.get_conn()
        
        def log_notice(msg):
            logging.info(f"PostgreSQL NOTICE: {msg}")
        
        conn.notice_handler = log_notice
        super().execute(context)

3. DAG中使用自定义Operator #

替换原PostgresOperator为LoggingPostgresOperator：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="postgres_log_example",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    sql_task = LoggingPostgresOperator(
        task_id="run_sql_with_logs",
        postgres_conn_id="your_postgres_conn",
        sql="/scripts/your_sql_file.sql"
    )

额外配置：连接参数优化 #

在Airflow的Postgres连接Extra字段添加以下内容，强制客户端消息级别：

{"options": "-c client_min_messages=notice"}

这样配置后，PostgreSQL的NOTICE消息（含受影响行数）会被Airflow任务日志捕获，全程无需用PythonOperator直接执行SQL。

内容的提问来源于stack exchange，提问作者AdamS