Airflow任务状态在Success与Removed间频繁切换问题求助

我部署了Airflow DAG，其中的任务状态持续在Success（成功）与Removed（已移除）之间反复切换，找不到任务状态从Success变为Removed的原因。

DAG代码 #

from airflow import DAG
import datetime
from datetime import timedelta
from tasks.user_space_tables_refresh import UserSpaceTablesRefresh

default_args = {
    'owner': 'Data Engineering',
    'depends_on_past': False,
    'start_date': datetime.datetime(2022, 2, 1),
}

user_space_dag = DAG(
    'user_space_snowflake_tables_refresh', default_args=default_args,
    schedule_interval="30 18 * * *", catchup=False)

with user_space_dag:
    users_task = UserSpaceTablesRefresh(
        task_id='ingest_USERS_data',
        source_table='USERS')

    saved_software_task = UserSpaceTablesRefresh(
        task_id='ingest_SAVED_SOFTWARE_data',
        source_table='SAVED_SOFTWARE')

任务类代码（tasks.user_space_tables_refresh） #

class UserSpaceTablesRefresh(BaseOperator):
    @apply_defaults
    def __init__(self, source_table, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.table = source_table

    def execute(self, context):
        try:
            sf_table = self.table
            ...

        except Exception as ex:
            print("Exception")

可能的原因及解决方法 #

1. DAG定义存在动态变更 #

如果DAG每次被Scheduler解析时，任务结构（比如task_id、依赖关系）发生变化，Airflow会判定旧任务已被移除，进而将已成功的任务标记为Removed。

• 检查代码中是否存在动态生成任务的逻辑，确保每次解析时任务列表、task_id完全一致。
• 确认UserSpaceTablesRefresh类的初始化逻辑没有引入动态变量，导致任务实例特征变化。

2. 元数据库一致性异常 #

Airflow元数据库（如PostgreSQL）中存储的任务状态与实际DAG定义不匹配，会触发状态回滚或标记为Removed。

• 执行airflow db check检查元数据库连接及完整性。
• 若确认元数据异常，可备份数据后执行airflow dags delete user_space_snowflake_tables_refresh清理旧运行记录，再重新部署DAG。

3. 任务执行逻辑存在缺陷 #

当前任务的execute方法仅打印异常却不抛出，可能导致状态判断混乱；另外若任务执行完成后存在未正确收尾的逻辑，也可能引发状态波动。

• 完善异常处理：在except块中添加raise ex，让Airflow正确识别失败状态。
• 确保execute方法执行完成后无残留的状态修改逻辑，保证任务状态上报准确。

4. Scheduler/Worker配置问题 #

Scheduler扫描DAG目录过于频繁，或Worker与Scheduler时间不同步，可能导致状态反复切换。

• 调整scheduler.cfg中的dag_dir_list_interval参数，避免过于频繁的DAG解析。
• 确认所有Airflow节点（Scheduler、Worker）的系统时间同步，时间差会引发状态判断错误。

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