如何在Airflow DAG中用Jinja嵌套模板传递任务名获取XCom值？

问题背景 #

现有可正常运行的S3KeySensor代码，能固定从aTestPyOperator任务拉取XCom值拼接bucket_key：

sensor = S3KeySensor(
        task_id='check_s3_for_file_in_s3',
        bucket_name='my_bucket',
        bucket_key= "{{params.folder}}/{{ ti.xcom_pull(task_ids='aTestPyOperator') }}/",
        params={'folder': 'test'}
    )

需求改为动态传入要拉取XCom的任务名，尝试两种写法均失败：

失败写法一（嵌套Jinja模板） #

sensor = S3KeySensor(
        task_id='check_s3_for_file_in_s3',
        bucket_name='my_bucket',
        bucket_key= "{{params.folder}}/{{ ti.xcom_pull(task_ids='{{params.fnName}}') }}/",
        params={'folder': 'test',
                'fnName': 'aTestPyOperator'}
    )

输出结果为test/NONE，原因是Jinja会把'{{params.fnName}}'当作字符串直接传入xcom_pull，而非解析成aTestPyOperator。

失败写法二（错误传递变量） #

bucket_key= "{{params.folder}}/{{ ti.xcom_pull(task_ids='params.fnName') }}/",

同样无效，因为这里把params.fnName当作字符串字面量传递，而非引用变量值。

解决方案 #

直接在Jinja模板中将params.fnName作为变量传入xcom_pull，不需要嵌套大括号或加引号：

sensor = S3KeySensor(
        task_id='check_s3_for_file_in_s3',
        bucket_name='my_bucket',
        bucket_key= "{{params.folder}}/{{ ti.xcom_pull(task_ids=params.fnName) }}/",
        params={'folder': 'test',
                'fnName': 'aTestPyOperator'}
    )

原理：Jinja模板中，params.fnName会被直接解析为对应参数值aTestPyOperator，作为xcom_pull的task_ids参数传入，从而正确拉取目标任务的XCom值。

替代实现方式 #

如果需要更灵活的逻辑，可以用PythonOperator生成bucket_key后通过XCom传递给S3KeySensor：

def generate_bucket_key(**context):
    folder = context['params']['folder']
    task_name = context['params']['fnName']
    xcom_value = context['ti'].xcom_pull(task_ids=task_name)
    return f"{folder}/{xcom_value}/"

generate_key_task = PythonOperator(
    task_id='generate_bucket_key',
    python_callable=generate_bucket_key,
    params={'folder': 'test', 'fnName': 'aTestPyOperator'},
    provide_context=True
)

sensor = S3KeySensor(
    task_id='check_s3_for_file_in_s3',
    bucket_name='my_bucket',
    bucket_key="{{ ti.xcom_pull(task_ids='generate_bucket_key') }}",
)

generate_key_task >> sensor

这种方式适合需要复杂逻辑处理的场景，比如对XCom值做额外转换、判断等。

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