如何将XCom作为Airflow PostgresOperator的参数传入？

我来帮你搞定这个问题——你现在的代码里犯了一个典型的Airflow模板语法混用错误，咱们一步步修正它：

问题根源 #

你在Python字典的params值里直接写了{{ int(ti.xcom_pull(task_ids='previous_task')) }}，这是Jinja模板的语法，但在纯Python代码层面这么写是无效的：Python会把它当成语法错误（双大括号不是合法的Python语法），就算你加了引号变成字符串，Airflow也不会自动渲染它，除非你明确告诉它这是需要模板处理的内容。

两种正确的实现方式 #

方式一：直接在SQL文件中使用XCom（推荐）

PostgresOperator的sql参数本身支持Jinja模板渲染，ti（任务实例）是模板上下文里默认可用的变量，所以你可以直接在SQL文件里获取XCom值，不需要在params里绕一圈：

1. 修改你的my_task.sql文件：

SELECT * FROM your_table 
WHERE some_column = {{ int(ti.xcom_pull(task_ids='previous_task')) }};

2. 简化PostgresOperator的代码：

my_task = PostgresOperator(
    task_id='my_task',
    postgres_conn_id=config.get(env, 'redshift_conn'),
    sql="my_task.sql",
    dag=dag
)

方式二：通过params传递XCom值

如果你一定要用params来传递参数，需要把参数值写成Jinja模板字符串，并且确保Airflow会渲染params（Airflow 2.x之后params默认支持模板渲染，1.x可能需要额外配置）：

1. 调整PostgresOperator的代码：

my_task = PostgresOperator(
    task_id='my_task',
    postgres_conn_id=config.get(env, 'redshift_conn'),
    sql="my_task.sql",
    params={
        # 把模板语法放在字符串里，让Airflow自动渲染
        'my_parameter': "{{ int(ti.xcom_pull(task_ids='previous_task')) }}"
    },
    # 如果你用的是Airflow 1.x，需要加上这行开启params的模板化
    # template_fields=('params',),
    dag=dag
)

2. 在my_task.sql里引用这个参数：

SELECT * FROM your_table 
WHERE some_column = {{ my_parameter }};

额外注意事项 #

• 确保previous_task确实推送了XCom值：如果是PythonOperator，它的返回值会自动推送到XCom；如果是其他算子，可能需要手动调用ti.xcom_push()。
• 如果previous_task返回的XCom本身就是整数，int()转换可以省略，但加上能避免字符串类型带来的问题。
• 如果你遇到XCom找不到的问题，可以检查task_ids是否拼写正确，或者是否设置了dag_id（跨DAG拉取XCom时需要）。

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