如何将Airflow XCom存储的值插入或更新到BigQuery表

你遇到的问题核心是混淆了DAG解析阶段和任务执行阶段：DAG文件加载时Airflow还没生成对应任务实例，自然无法调用xcom_pull拉取运行时才产生的XCom值，所有依赖XCom的动态逻辑都要放到任务执行阶段处理，以下是三种常用算子的具体实现方式：

BigQuery官方算子的sql、query_params参数原生支持Jinja模板渲染，你不需要提前在DAG解析阶段拼接SQL，直接在参数里写入XCom的模板引用即可，算子执行时会自动拉取对应XCom值完成替换。
示例代码：

from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator
# 高版本Google Provider可替换为BigQueryInsertJobOperator，配置逻辑一致

# 假设你存储db_value的上游任务id为upstream_db_extract
write_to_bq = BigQueryExecuteQueryOperator(
    task_id='write_xcom_to_bq',
    sql="""
        -- 插入场景
        INSERT INTO `your_project.your_dataset.target_table` (col1, col2)
        VALUES ('fixed_value', '{{ ti.xcom_pull(task_ids="upstream_db_extract", key="db_value") }}');
        
        -- 更新场景直接替换为对应UPDATE语句即可
        -- UPDATE `your_project.your_dataset.target_table` SET target_col = '{{ ti.xcom_pull(task_ids="upstream_db_extract", key="db_value") }}' WHERE filter_col = 'fixed_condition';
    """,
    use_legacy_sql=False,
    gcp_conn_id='your_gcp_connection_id'
)

注意：如果db_value是数值、布尔类非字符串类型，去掉值包裹的单引号即可；如果值本身包含特殊字符，推荐用参数化查询避免SQL注入和格式错误：

write_to_bq_safe = BigQueryExecuteQueryOperator(
    task_id='write_xcom_safe',
    sql="INSERT INTO `your_project.your_dataset.target_table` (value_col) VALUES (%(db_val)s)",
    query_params=[
        ('db_val', 'STRING', "{{ ti.xcom_pull(task_ids='upstream_db_extract', key='db_value') }}")
    ],
    use_legacy_sql=False,
    gcp_conn_id='your_gcp_connection_id'
)

如果你需要在写入BigQuery前对XCom值做校验、格式转换等额外处理，可以把拉取XCom、拼接SQL、执行查询的逻辑全部写到Python callable内部——callable是任务执行时才会运行的，这时候任务实例上下文已经生成，可以正常调用xcom_pull。
示例代码：

from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
from airflow.operators.python import PythonOperator

def bq_write_callable(**context):
    # 执行阶段从上下文拉取XCom
    db_value = context['ti'].xcom_pull(task_ids='upstream_db_extract', key='db_value')
    # 此处可加任意自定义处理逻辑，比如空值校验、格式转换
    processed_value = str(db_value).strip()
    # 拼接SQL
    insert_sql = f"INSERT INTO `your_project.your_dataset.target_table` (value_col) VALUES ('{processed_value}')"
    # 调用BigQuery Hook执行SQL
    bq_hook = BigQueryHook(gcp_conn_id='your_gcp_connection_id', use_legacy_sql=False)
    bq_hook.run(insert_sql)

write_via_python = PythonOperator(
    task_id='write_via_python',
    python_callable=bq_write_callable
    # Airflow 2.x无需手动设置provide_context=True，默认自动传入上下文
)

如果你习惯用命令行操作，BashOperator的bash_command参数同样支持Jinja模板渲染，可以直接在命令中嵌入XCom值调用bq客户端执行写入：

from airflow.operators.bash import BashOperator

write_via_bash = BashOperator(
    task_id='write_via_bash',
    bash_command="""
        bq query --use_legacy_sql=false \
        "INSERT INTO \`your_project.your_dataset.target_table\` (value_col) VALUES ('{{ ti.xcom_pull(task_ids='upstream_db_extract', key='db_value') }}')"
    """
    # 提前在Worker节点配置好GCP认证环境变量，或在env参数中传入服务账号密钥路径
)

• 严禁在DAG文件顶层、算子初始化的非模板参数位置直接调用xcom_pull，这个阶段属于DAG解析期，没有运行时任务实例上下文，一定会拉取值失败或报错。
• 使用Jinja模板传参时，要确认对应算子的参数在template_fields列表内，BigQuery算子的sql/query_params、BashOperator的bash_command默认都在列表内，可以直接渲染；PythonOperator如果要在初始化参数中用模板，需要手动把对应字段加入template_fields，否则不会渲染，更稳妥的方式是直接在callable内从上下文拉取XCom。
• 如果XCom存储的是字典、列表这类复杂结构，可以在Jinja模板中加| tojson过滤器做序列化，避免拼接SQL时出现格式错误。
• 使用BashOperator方案前，要确认Worker节点已经安装Google Cloud SDK并配置了对应BigQuery操作权限，否则会出现命令不存在或权限不足的错误。

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