如何让Airflow实现Luigi同类特性：避免已输出有效结果的未修改任务重跑

Airflow完全支持你需要的「已成功、未修改任务重跑自动跳过」能力，常用实现方式分两类：

仅校验输出存在性的轻量方案 #

如果你的场景只需要判断任务输出未丢失、不需要校验代码变更，可以直接在任务入口加输出检查逻辑，符合要求就直接标记任务成功。你可以封装通用装饰器复用逻辑：

from airflow.utils.state import State
import os

def skip_if_valid_output(func):
    def wrapper(*args, **kwargs):
        ti = kwargs['ti']
        # 替换为你的任务输出校验逻辑，比如检查文件存在、大小符合预期等
        task_output = get_task_output_path(ti.task_id, ti.execution_date)
        if os.path.exists(task_output):
            ti.set_state(State.SUCCESS)
            return
        return func(*args, **kwargs)
    return wrapper

# 给对应任务加装饰器即可
@skip_if_valid_output
@task
def your_data_processing_task():
    # 原有任务逻辑
    pass

同时校验代码未变更的完整方案 #

如果需要和Luigi一样，只有任务输出存在+任务代码未修改才跳过，可以在上述逻辑基础上增加代码哈希校验：

1. 任务首次执行成功时，将当前任务代码的哈希值写入输出文件元数据、或存入XCom
2. 后续重跑时，先比对当前任务代码哈希和上次成功存储的哈希值，同时校验输出有效性
3. 两个条件都满足则直接标记成功，否则正常执行任务

原生Task Caching方案（Airflow 2.4+） #

2.4及以上版本的Airflow提供了原生任务缓存能力，不需要自己写校验逻辑，只需给任务配置cache_key参数即可：

# TaskFlow API写法
@task(
    # cache_key可自定义，这里用任务ID+代码哈希作为唯一标识
    cache_key="{{ ti.task_id }}_{{ hash(ti.task_code) }}",
    # 缓存有效期，按需设置
    cache_timeout=86400 * 30
)
def your_task():
    # 原有任务逻辑
    pass

# 传统Operator写法
PythonOperator(
    task_id="your_task",
    python_callable=your_callable,
    cache_key="{{ ti.task_id }}_{{ hash(ti.task_code) }}",
    cache_timeout=86400 * 30
)

配置完成后，只要cache_key匹配且存在对应成功执行记录，Airflow会自动跳过该任务，复用之前的执行结果。

另外你可以配合DAG的catchup=False、任务的depends_on_past=True参数，进一步减少不必要的重复执行。

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