Airflow：如何从上游任务获取模板字段用于S3Sensor

我来帮你梳理下当前的问题，以及对应的解决方案：

问题根源 #

你当前在S3KeySensor的bucket_key里用{{ snapshot_date_str }}，这个写法是直接尝试引用全局模板上下文变量，但你的snapshot_date_str是通过上游ContextInitOperator的XCom推送的，并不是全局模板变量，所以Airflow没办法直接识别到它。

解决方案：使用XCom模板宏获取上游变量 #

Airflow提供了ti（Task Instance的缩写）模板变量，通过它可以调用xcom_pull()方法来获取上游任务推送的XCom值。你只需要修改bucket_key的模板写法即可：

file_task = S3KeySensor(
    task_id="s3_file_sensor",
    aws_conn_id='s3_connection',
    # 使用ti.xcom_pull获取上游任务的XCom值
    bucket_key='test/{{ ti.xcom_pull(task_ids="set_context", key="snapshot_date_str") }}/abc.csv',
    bucket_name=S3_BUCKET,
    wildcard_match=True,
    poke_interval=10,
    timeout=60,
    dag=dag
)

额外的优化点 #

看你的ContextInitOperator代码，template_fields = ('snapshot_date_str',)其实是多余的——因为这个字段是在__init__方法里直接用snapshot_date生成的，不需要Airflow对它进行模板渲染。你可以去掉这一行，避免不必要的模板处理：

class ContextInitOperator(BaseOperator):
    # 去掉多余的template_fields
    @apply_defaults
    def __init__(
            self, snapshot_date, *args, **kwargs):
        super(ContextInitOperator, self).__init__(*args, **kwargs)
        self.snapshot_date_str = snapshot_date.strftime('%Y-%m-%d')
    
    def execute(self, context):
        context['task_instance'].xcom_push(key='snapshot_date_str', value=self.snapshot_date_str)

验证要点 #

• 确保上游set_context任务成功执行，你可以在Airflow UI的任务实例详情里查看XCom是否被正确推送
• 确认S3KeySensor的bucket_key模板语法正确，Airflow会在任务运行时自动渲染这个模板，替换成对应的日期字符串

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