Airflow无法识别自定义Helper类生成的DAG问题求助

根因1：缺失DAG类导入 #

你的Helper类代码中使用了DAG类构建实例，但顶部导入语句仅引入了ScheduleInterval，未导入DAG类：

# 你的现有导入
from airflow.models.dag import ScheduleInterval
# 缺少以下导入
from airflow.models.dag import DAG

部分情况下本地测试时如果其他代码提前导入过DAG不会报错，但Airflow调度器独立解析DAG文件时会抛出导入异常，直接跳过该DAG的加载。

根因2：start_date使用动态值 #

你在默认参数中用datetime.now()-timedelta(days=1)作为start_date，这是Airflow加载DAG的典型禁忌：

• Airflow调度器每次扫描DAG文件都会重新计算这个值，导致DAG的调度基准时间不断变化
• 调度器会判定这类DAG的调度逻辑不稳定，直接跳过加载避免状态混乱

根因3：模块导入路径异常 #

你自定义的dagClass模块如果不在Airflow运行环境的PYTHONPATH范围内，调度器解析DAG文件时会抛出ImportError，导致DAG加载失败：

• 本地测试时的Python环境和Airflow调度器运行的环境可能存在差异
• 你可以通过airflow dags list-import-errors命令直接查看所有DAG的导入报错信息

修复步骤 #

• 补全Helper类的DAG导入语句
• 将start_date修改为固定静态值，示例：

self.args={
    # 其他参数不变
    'start_date': datetime(2024, 1, 1), # 替换为你需要的固定起始时间
}

• 将dagClass.py放到Airflow的dags根目录下，或者将其所在路径添加到Airflow运行环境的PYTHONPATH中
• 执行airflow dags list验证DAG是否被正常识别

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