如何在Snowflake转Google Sheet的SQL WHERE子句中正确使用Airflow xcom_pull

AI 小北

2026-8-21

解决Airflow SQL WHERE子句匹配XCom字符串值的问题

修正XCom拉取语法，指定key并正确包裹引号
你在ti.xcom_pull时未指定key="run_id"，但存储时用的是xcom_push(key="run_id")，默认拉取的return_value可能为None；同时字符串类型的run_id在SQL中需要被单引号包裹。正确的WHERE子句写法：
```
WHERE RUN_ID = '{{ ti.xcom_pull(task_ids="Task-2", key="run_id") }}'
```
注意task_ids用双引号，避免和外层单引号冲突。

验证拉取的run_id是否正确
可临时添加打印任务，确认XCom值是否被正确获取：

def check_run_id(**context):
    ti = context['ti']
    run_id = ti.xcom_pull(task_ids='Task-2', key='run_id')
    print(f"获取到的run_id: {run_id}，类型: {type(run_id)}")

check_task = PythonOperator(
    task_id='Check-Run-ID',
    python_callable=check_run_id,
    provide_context=True,
    dag=dag
)

运行后查看该任务日志，确认run_id的内容和类型是否符合预期。

确认Snowflake表字段类型匹配
确保Snowflake中RUN_ID列是字符串类型（如VARCHAR），和你生成的UUID字符串类型一致，避免因类型不匹配导致过滤无结果。
引号冲突的替代写法
如果SQL语句用双引号包裹，内部可使用转义单引号：
```
"WHERE RUN_ID = '{{ ti.xcom_pull(task_ids=''Task-2'', key=''run_id'') }}'"
```