Airflow如何不依赖context获取上一次DAG运行推送的XCom值

方案说明 #

首先明确核心限制：你无法在__init__方法中获取到对应XCom值，原因是__init__属于DAG解析阶段执行的逻辑，此时还没有实际的DAG运行实例，XCom是任务运行时产生的动态数据，仅能在任务执行阶段访问。
同时不建议在__init__中加入任何访问元数据的逻辑，会导致DAG解析效率下降，严重时会引起Scheduler负载过高、DAG加载失败。

替代实现方案 #

你可以将获取上一次运行XCom的逻辑封装为Sensor类的内部方法，在poke方法中首次调用时执行查询，无需依赖传入的context对象，也不需要直接操作底层数据库，直接用Airflow官方提供的模型类方法即可：

from airflow.models import TaskInstance, XCom
from airflow.utils.state import State

class YourS3Sensor(BaseSensorOperator):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 不要在这里加查询XCom的逻辑
        
    def _get_previous_files(self):
        # 封装获取历史文件列表的逻辑，无需依赖context
        prev_ti = TaskInstance.get_previous_ti(
            dag_id=self.dag_id,
            task_id=self.task_id,
            state=State.SUCCESS
        )
        if not prev_ti:
            # 首次运行没有历史数据
            return []
        return XCom.pull(
            key='new_files',
            task_ids=self.task_id,
            dag_id=self.dag_id,
            execution_date=prev_ti.execution_date
        )
    
    def poke(self, context):
        previous_files = self._get_previous_files()
        # 后续扫描S3的逻辑
        ...

如果需要在一次任务的多次poke调用中复用查询结果，可以给方法加缓存装饰器避免重复查询：

from functools import lru_cache

class YourS3Sensor(BaseSensorOperator):
    # ... 其他逻辑不变
    @lru_cache(maxsize=None)
    def _get_previous_files(self):
        # 同上查询逻辑

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