Airflow中BigQueryInsertJobOperator实现XCom值推送的问题咨询

错误根因 #

你的代码存在两个核心问题导致报错和无法推送XCom：

1. 执行时机错误：你将task_instance.xcom_push直接写在了BigQueryInsertJobOperator的入参位置，这段代码会在DAG解析阶段就运行，此时既没有运行时生成的task_instance对象，也还没执行BQ查询拿到yr_wk的值，Airflow会误将未定义的task_instance识别为Airflow变量，因此抛出变量不存在的报错。
2. 逻辑结构错误：你在PythonOperator的回调函数里仅定义了BQ Operator，但没有实际执行查询获取返回值，自然取不到要推送的yr_wk。

解决方案 #

直接在PythonOperator的回调函数内使用BigQueryHook执行查询，拿到结果后再推送XCom，比嵌套BQ Operator更简洁高效，代码实现如下：

第一步：调整任务1代码 #

# 导入BQ Hook
from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook

def get_bq_wm_yr_wk(**kwargs):
    # 从运行时上下文获取task_instance对象
    ti = kwargs["ti"]
    # 初始化BQ Hook，使用你已配置的GCP连接ID
    bq_hook = BigQueryHook(gcp_conn_id=gcp_connection_id, use_legacy_sql=False)
    # 执行查询获取结果，get_first返回查询的第一行结果
    query_result = bq_hook.get_first(sql=bq_query)
    yr_wk = query_result[0]
    # 推送值到XCom
    ti.xcom_push(key="yr_wk", value=yr_wk)
    # 打印日志方便直接在任务日志中查看取值
    print(f"查询获取的yr_wk值为：{yr_wk}")

# 任务1定义保持基本不变，Airflow 2.x无需额外配置provide_context，默认传递上下文
get_wm_yr_wk = PythonOperator(
    task_id="get_wm_yr_wk",
    python_callable=get_bq_wm_yr_wk,
    provide_context=True,
    on_failure_callback=failure_callback,
    on_retry_callback=failure_callback,
    dag=dag
)

第二步：验证XCom推送结果 #

任务1运行成功后，有两种方式查看yr_wk取值：

• 进入任务详情页，点击XCom标签页，即可看到你推送的key为yr_wk的记录
• 直接查看任务运行日志，找到你打印的yr_wk取值日志即可

任务2使用XCom值示例 #

如果你的第二个任务使用BigQueryInsertJobOperator，可以直接在查询SQL中通过jinja模板引用XCom值：

-- 你的第二个任务SQL示例，假设存为sql文件
SELECT * FROM 你的业务表 WHERE yr_wk = {{ ti.xcom_pull(task_ids='get_wm_yr_wk', key='yr_wk') }}

定义第二个任务时，确保SQL字段开启模板渲染（BigQuery官方Operator默认支持SQL的模板渲染，无需额外配置），即可自动读取任务1推送的yr_wk值运行查询，后续按原有逻辑导出到GCS即可。

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