Airflow任务重跑时如何无DB/文件持久化类执行状态字典？

核心方案 #

利用Airflow内置的XCom机制存储类的执行状态——XCom是Airflow原生用于任务内/任务间传递小数据的组件，默认依托Airflow元数据库存储，无需额外搭建外部数据库或文件，完全满足你"不使用数据库或文件"的要求（这里的元数据库是Airflow自身运维依赖，不属于额外需要你维护的外部存储）。

具体实现步骤 #

• 拆分执行单元：把单个任务中同时处理3个类的逻辑，拆分为每个类的独立执行块，便于逐个判断状态。
• 读取历史状态：任务启动时，先从XCom拉取当前任务实例对应的类执行状态字典，首次执行则初始化空字典。
• 断点续跑逻辑：遍历待处理类列表，仅执行未标记为成功的类；每成功处理一个类，就更新状态字典并推送到XCom持久化。
• 失败重跑兼容：若某类执行失败，任务会被标记为失败；重跑时任务会读取XCom中存储的状态，跳过已成功的类，直接从失败的类开始执行。

代码示例 #

from airflow.decorators import task
from airflow.models import TaskInstance
from airflow.utils.session import provide_session

@task
def process_classes(task_instance: TaskInstance, **context):
    # 当前任务负责处理的类列表
    target_classes = ["B", "C", "D"]
    # 从XCom获取已执行状态，无历史数据则初始化空字典
    exec_status = task_instance.xcom_pull(task_ids=context["task"].task_id, key="class_exec_status") or {}
    
    for cls in target_classes:
        # 跳过已成功执行的类
        if exec_status.get(cls) == "Y":
            print(f"跳过已完成的类: {cls}")
            continue
        
        try:
            # 替换为你的实际类处理逻辑（比如更新数据表）
            print(f"开始处理类: {cls}")
            # process_class_logic(cls)  # 实际业务代码
            
            # 执行成功，更新状态并推送到XCom
            exec_status[cls] = "Y"
            task_instance.xcom_push(key="class_exec_status", value=exec_status)
            print(f"类 {cls} 处理完成，状态已保存")
        except Exception as e:
            print(f"类 {cls} 处理失败: {str(e)}")
            raise e  # 抛出异常，标记任务失败

关键注意点 #

• XCom适合存储小体积数据（默认限制48KB），仅存状态字典完全符合要求。
• 确保任务task_id唯一，XCom按task_id和dag_run_id关联存储，不会和其他任务的状态混淆。
• 重跑时选择"重跑失败的任务实例"，Airflow会保留对应实例的XCom数据，保证状态读取准确。

内容的提问来源于stack exchange，提问作者A gupta