如何用BigQueryInsertJobOperator向BigQuery存储过程传递XCOM参数

错误原因 #

你之前的写法存在两个核心问题：

1. 混淆了Airflow的params参数与BigQuery API的参数配置：Airflow的params用于SQL模板渲染，而非直接传递给BigQuery的参数列表；且你直接将函数赋值给params，但它需要的是字典类型。
2. BigQuery的parameters配置仅定义了参数类型，未传入实际参数值，导致SQL中的@p_fecha_ini找不到对应值。

正确解决方案 #

以下提供两种可行的实现方式：

方式一：使用BigQuery原生参数化查询（推荐，支持复杂类型） #

先通过PythonOperator获取并缓存参数，再在BigQueryInsertJobOperator中用Jinja模板注入参数值：

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

def get_params_dict(**context):
    table_rows = context['ti'].xcom_pull(task_ids='get_data_from_table')
    # 取第一行的第3个元素（索引2），根据实际数据结构调整
    if table_rows:
        return {"p_fecha_ini": table_rows[0][2]}
    # 无数据时返回默认值
    return {"p_fecha_ini": "1970-01-01"}

# 新增任务：获取参数并存入XCOM
get_params_task = PythonOperator(
    task_id='get_params_task',
    python_callable=get_params_dict,
    provide_context=True,
    do_xcom_push=True
)

# 调用存储过程的任务
start_sp = BigQueryInsertJobOperator(
    task_id='start_sp',
    configuration={
        'query': {
            'query': f"CALL `{PROJECT_ID}.{DATASET_ID}.{SP_NAME}`(@p_fecha_ini)",
            'useLegacySql': False,
            'parameters': [
                {
                    'name': 'p_fecha_ini',
                    'parameterType': {'type': 'STRING'},
                    'parameterValue': {'value': "{{ ti.xcom_pull(task_ids='get_params_task')['p_fecha_ini'] }}"}
                }
            ]
        }
    },
    trigger_rule='all_done',
    # 开启configuration字段的Jinja模板渲染
    template_fields=['configuration']
)

# 设置任务依赖
get_params_task >> start_sp

方式二：直接用Jinja模板替换SQL参数（简单场景） #

如果参数是基础类型（如字符串、日期），可直接将XCOM值渲染到SQL语句中：

start_sp = BigQueryInsertJobOperator(
    task_id='start_sp',
    configuration={
        'query': {
            # 注意字符串参数需包裹单引号，日期类型需符合BigQuery格式
            'query': f"CALL `{PROJECT_ID}.{DATASET_ID}.{SP_NAME}`('{{ ti.xcom_pull(task_ids='get_data_from_table')[0][2] }}')",
            'useLegacySql': False,
        }
    },
    trigger_rule='all_done',
    template_fields=['configuration']
)

关键注意事项 #

• 确保get_data_from_table任务返回的XCOM数据结构符合预期（你当前代码假设返回多行，取每行的第3个元素，但最终只保留最后一行的值，需根据实际业务调整）。
• 若参数是数值、日期等非字符串类型，需调整parameterType及SQL中的格式（如日期无需单引号，数值直接传入）。

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