如何将XCom共享变量作为BigQueryInsertJobOperator配置的键？

核心问题 #

Airflow的Jinja2模板渲染仅作用于字典的值，不会自动解析字典的键。你把模板语法写在字典的键位置时，Airflow不会将其替换为实际的mode值，而是直接把{{task_instance.xcom_pull(task_ids='capture_mode', key='mode')}}当作字符串键去查找，自然找不到对应值，触发KeyError。

解决方法 #

方法1：在PythonOperator中直接生成完整tableId #

既然mode是在PythonOperator中生成的，可直接在此任务中组装好BigQuery需要的tableId，再推送到XCom，后续任务直接引用完整值：

def capture_mode(**context):
    mode = context['dag_run'].conf.get('mode')
    # 定义mode与tableId的映射关系
    table_id_map = {
        'dev': 'project.dataset.dev_table',
        'prod': 'project.dataset.prod_table'
    }
    target_table_id = table_id_map[mode]
    # 推送完整tableId到XCom
    context['task_instance'].xcom_push(key='target_table_id', value=target_table_id)

capture_mode_task = PythonOperator(
    task_id='capture_mode',
    python_callable=capture_mode,
    provide_context=True,
    dag=dag
)

bq_task = BigQueryInsertJobOperator(
    task_id='run_bq_job',
    configuration={
        'query': {
            'query': 'SELECT * FROM `{{task_instance.xcom_pull(task_ids="capture_mode", key="target_table_id")}}`',
            'useLegacySql': False
        }
    },
    dag=dag
)

方法2：推送完整的BigQuery配置到XCom #

如果需要动态生成整个BigQuery任务配置，可在PythonOperator中组装好完整的configuration，再推送到XCom，后续任务直接引用：

def capture_mode(**context):
    mode = context['dag_run'].conf.get('mode')
    table_id_map = {
        'dev': 'project.dataset.dev_table',
        'prod': 'project.dataset.prod_table'
    }
    # 组装完整的BigQuery任务配置
    bq_config = {
        'query': {
            'query': f'SELECT * FROM `{table_id_map[mode]}`',
            'useLegacySql': False
        }
    }
    context['task_instance'].xcom_push(key='bq_config', value=bq_config)

bq_task = BigQueryInsertJobOperator(
    task_id='run_bq_job',
    configuration="{{task_instance.xcom_pull(task_ids='capture_mode', key='bq_config')}}",
    dag=dag
)

方法3：直接在模板值中拼接tableId #

如果tableId有固定规律（比如前缀+mode+后缀），可直接在tableId的模板值里动态拼接mode，无需字典映射：

bq_task = BigQueryInsertJobOperator(
    task_id='run_bq_job',
    configuration={
        'query': {
            'query': 'SELECT * FROM `project.dataset.{{task_instance.xcom_pull(task_ids="capture_mode", key="mode")}}_table`',
            'useLegacySql': False
        }
    },
    dag=dag
)

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