Airflow：BigQueryInsertJobOperator无法解析XCom传递的参数

问题根源 #

BigQueryInsertJobOperator的params参数默认不在Airflow的模板渲染字段列表中，因此你写在params里的Jinja表达式{{ task_instance.xcom_pull(...) }}不会被解析，直接以原始字符串传递给SQL模板，最终导致存储过程接收到的是未渲染的模板字符串而非实际XCom值。

解决方法 #

方法1：直接在SQL文件中拉取XCom值（最简单）

不需要通过params中转，直接在SQL文件里使用Jinja表达式拉取XCom值：

CALL `project.db.storedproc`('{{ task_instance.xcom_pull(task_ids="func1", key="param1") }}');

对应的BigQueryInsertJobOperator可以移除params参数：

bq_op_load = BigQueryInsertJobOperator(
    task_id="bq_op_load",
    configuration={
        "query": {
            "query": "{% include 'sql_file_path.sql' %}",
            "useLegacySql": False,
        }
    },
    dag=dag
)

方法2：让params支持模板渲染

自定义一个继承自BigQueryInsertJobOperator的子类，将params加入模板渲染字段：

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

class TemplatedBigQueryInsertJobOperator(BigQueryInsertJobOperator):
    # 将params添加到模板字段列表，使其支持Jinja渲染
    template_fields = (*BigQueryInsertJobOperator.template_fields, "params")

然后使用这个自定义Operator，原有的params配置就能正常解析Jinja表达式了：

bq_op_load = TemplatedBigQueryInsertJobOperator(
    task_id="bq_op_load",
    configuration={
        "query": {
            "query": "{% include 'sql_file_path.sql' %}",
            "useLegacySql": False,
        }
    },
    params={
        "param1": "{{ task_instance.xcom_pull(task_ids='func1', key='param1') }}"
    },
    dag=dag
)

额外优化（Airflow 2.x适配）

PythonOperator的provide_context=True在Airflow 2.x已被废弃，建议直接通过参数接收ti（TaskInstance）对象：

def func1(ti):
    ti.xcom_push(key='param1', value=param1)

func1 = PythonOperator(
    task_id="func1",
    python_callable=func1,
    dag=dag
)

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