Airflow DAG执行异常：任务处于成功状态无法重新运行

阿华AIGC实验室

2026-5-13

解决Airflow DAG触发后任务处于Success状态无法执行的问题

这个问题我碰到过好多次，核心原因其实很直白：Airflow默认不会重新执行已经处于success状态的任务实例。首次运行时所有任务都是全新的，没有历史状态，所以能正常跑起来，但之后再触发DAG时，Airflow检测到任务已经成功完成过，就会跳过执行直接标记为成功，这就是你看到的报错：

dependency 'Task Instance State' FAILED: Task is in the 'success' state which is not a valid state for execution. The task must be cleared in order to be run

下面给你几个实用的解决思路，从临时救急到长期优化都覆盖到了：

一、临时快速解决：清理任务实例状态

如果只是想让任务立刻重新执行，直接重置之前的成功状态就行：

UI操作：打开Airflow UI，找到目标DAG（example_trigger_target_dag），进入「Graph View」或「Tree View」，选中对应的run_this任务，点击顶部的「Clear」按钮确认重置，之后再触发控制器DAG就能正常执行任务了。
CLI命令：习惯用命令行的话，Airflow 2.x可以用以下命令清理整个DAG的任务状态：
```
airflow dags clear example_trigger_target_dag
```
或者只清理特定任务：
```
airflow tasks clear -d example_trigger_target_dag -t run_this
```

二、长期优化：修改DAG配置避免状态冲突

1. 给TriggerDagRunOperator添加`reset_dag_run=True`

在控制器DAG的TriggerDagRunOperator里加上这个参数，每次触发目标DAG时会自动重置之前的DAG Run状态，确保任务能重新执行：

trigger = TriggerDagRunOperator(
    task_id='test_trigger_dagrun',
    trigger_dag_id="example_trigger_target_dag",
    python_callable=conditionally_trigger,
    params={'condition_param': True, 'message': 'Hello World'},
    reset_dag_run=True,  # 新增这个参数
    dag=dag,
)

2. 给目标DAG生成唯一Run ID

如果希望每次触发都是独立的DAG Run，不会复用之前的状态，可以在conditionally_trigger函数里手动指定唯一的run_id，比如用时间戳：

def conditionally_trigger(context, dag_run_obj):
    c_p = context['params']['condition_param']
    if c_p:
        dag_run_obj.payload = {'message': context['params']['message']}
        # 生成唯一run_id，避免复用旧的运行实例
        dag_run_obj.run_id = f"triggered__{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
        pp.pprint(dag_run_obj.payload)
    return dag_run_obj

3. 修正DAG的`start_date`为固定值

你的两个DAG都用了datetime.utcnow()作为start_date，这是不推荐的——每次Airflow解析DAG时，utcnow()都会生成新时间，导致Airflow认为这是新的DAG版本，可能引发奇怪的状态问题。建议改成固定的过去时间：

# 在trigger_controller_dag.py里
default_args={
    "owner": "airflow",
    "start_date": datetime(2023, 1, 1),  # 改成固定日期
},

# 在trigger_target_dag.py里
args = {
    'start_date': datetime(2023, 1, 1),  # 同样改成固定日期
    'owner': 'airflow',
}