如何在Airflow中将PostgreSQL查询结果存入变量（Postgres Operator/Hook）

1. 现有代码XCom推送逻辑说明 #

你当前写的_query_postgres函数已经完成了XCom数据推送，有两个可用的XCom键可以获取结果：

• 你显式调用xcom_push指定的my_value键
• PythonOperator默认会将函数返回值自动推送到return_value键，两个键存储的内容完全一致

2. 下游BashOperator获取结果的方法 #

你可以直接在BashOperator的bash_command参数中使用Jinja模板语法拉取XCom值，示例代码如下：

from airflow.operators.bash import BashOperator

# 先定义查询Postgres的PythonOperator
query_postgres_task = PythonOperator(
    task_id="query_postgres_task",
    python_callable=_query_postgres,
    # *如果是Airflow 1.x需要显式配置provide_context=True，Airflow 2.x版本默认开启上下文传递不需要配置
    provide_context=True,
    dag=dag
)

# 下游Spark任务的BashOperator
spark_task = BashOperator(
    task_id="run_spark_job",
    bash_command="""
        spark-submit \
        --name your_spark_job \
        your_spark_script.py \
        --pg-result '{{ ti.xcom_pull(task_ids="query_postgres_task", key="my_value") }}'
    """,
    dag=dag
)

# 定义依赖
query_postgres_task >> spark_task

如果需要在Bash脚本中解析JSON格式的查询结果，可以使用jq工具处理，示例：

# 拿查询结果的第一行第二个字段
echo '{{ ti.xcom_pull(task_ids="query_postgres_task", key="my_value") }}' | jq '.[0][1]'

注意：Airflow XCom默认存储在元数据库中，不适合传递过大的查询结果。如果结果数据量超过10MB，建议将结果写入对象存储/本地文件，仅将文件路径推入XCom传递给下游。

3. PostgresOperator返回None的原因 #

PostgresOperator的设计定位是执行DDL、DML等不需要返回结果的SQL语句，默认不会将查询结果返回或推入XCom，所以你拿到None是预期行为。如果需要获取查询结果，推荐继续使用当前的PostgresHook+PythonOperator的实现方式。

现有代码优化建议 #

你代码中cursor.execute的返回值为None，赋值的mark_williams变量无实际用途可删除；另外建议显式调用cursor.fetchall()获取结果，可读性更好：

# 原代码的details_dicts = [doc for doc in cursor] 替换为
details_dicts = cursor.fetchall()

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