Airflow动态传参时Xcom值无法正确传入SQL where子句问题如何解决？

你遇到的问题本质是Python字符串格式化的转义规则和Airflow Jinja模板的语法产生了冲突：

• Python的str.format()方法会将字符串中连续的两个大括号{{转义为单个{，你原始写法中包裹ti.xcom_pull的双层大括号经过format处理后会变成单层大括号，Airflow无法识别为需要渲染的Jinja模板，因此直接将模板原文输出到了SQL中。
• 额外注意你代码中存在task_id拼写不一致的潜在问题：部分代码中task_id为Match_Updated_date_{}（单数date），部分为Match_Updated_dates_{}（复数dates），拼写错误会导致Xcom拉取失败。

提供两种可直接落地的实现方式，推荐优先使用第一种，无转义风险：

方案1：通过params参数传递动态值（推荐） #

利用PostgresOperator原生支持的params参数传递country变量，直接在SQL模板中完成Jinja语法拼接，完全避免Python层格式化的转义问题：

import_redshift_table_zm = PostgresOperator(
    task_id='copy_data_from_redshift_zm',
    postgres_conn_id='postgres_default',
    # 把动态变量country传入模板上下文
    params={"country": country},
    sql="""
        BEGIN;
        create table angaza_public_spark.stag_angaza_users_zm as
        SELECT * FROM angaza_public_zm.users
        -- 直接在Jinja模板中拼接动态参数，~是Jinja的字符串拼接运算符
        where last_updated_utc > '{{ ti.xcom_pull(task_ids="Match_Updated_dates_" ~ params.country, key="QueryTimeStamp_" ~ params.country) }}';
        END;
    """
)

方案2：调整大括号数量适配format转义规则 #

如果你坚持用Python的format方法拼接模板，需要写四层大括号，保证经过format转义后仍保留Jinja需要的双层大括号：

# 四层大括号经过format后会转为两层，符合Jinja模板语法要求
xcom_template = "{{{{ ti.xcom_pull(task_ids='Match_Updated_dates_{0}', key='QueryTimeStamp_{0}') }}}}".format(country)

import_redshift_table_zm = PostgresOperator(
    task_id='copy_data_from_redshift_zm',
    postgres_conn_id='postgres_default',
    sql="""
        BEGIN;
        create table angaza_public_spark.stag_angaza_users_zm as
        SELECT * FROM angaza_public_zm.users
        where last_updated_utc > '{}';
        END;
    """.format(xcom_template)
)

• 确保你推送Xcom的Python函数中已经从kwargs中提取了ti对象，你提供的代码片段中缺失这一步，会导致Xcom推送失败：

def match_dates(**kwargs):
    # 必须先提取ti对象
    ti = kwargs['ti']
    try:
        print("enters the try block")
        response = s3.get_object(Bucket='mygluecrawlerbucket',Key='DateTime/Users/my_date_{}.txt'.format(kwargs['key1']))
        print("response is ", response)
        status = response['ResponseMetadata']['HTTPStatusCode']
        if status == 200:
            print("Enters the status block ")
            data = response['Body'].read().decode("utf-8")
            ti.xcom_push(key="QueryTimeStamp_{}".format(country), value=data)

• 必须设置任务依赖，保证Xcom推送任务在PostgresOperator之前执行：

# 假设推送Xcom的任务id为match_dates_task
match_dates_task >> import_redshift_table_zm

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