Airflow拆分DAG生成逻辑到独立文件后Web UI无显示问题咨询

• 导入拼写错误
  你存放生成函数的文件名为dynamic_dags.py，但load_dags.py中的导入语句写为from dynamic_dag import generate_dag，少了末尾的s，直接触发导入错误，DAG生成逻辑无法运行。
• 依赖和变量定义缺失
  你提供的dynamic_dags.py代码中没有导入DAG、DummyOperator、days_ago等依赖类和方法，也没有定义default_args用到的args变量，拆分到独立文件后函数运行会触发NameError，Airflow会跳过加载存在运行错误的文件。
• 文件存放不符合扫描规则
  Airflow默认只扫描airflow.cfg中dags_folder配置路径下的顶层Python文件，以及包含__init__.py的子目录下的Python文件。如果load_dags.py不在扫描路径内，或是dynamic_dags.py所在目录没有__init__.py无法被导入，都会导致DAG加载失败。
• 全局作用域暴露失败
  虽然你使用globals()注册DAG对象，但若存在前面的导入、运行错误，这行注册代码不会执行，自然没有符合要求的全局DAG对象被Airflow识别。

修复方案 #

1. 修正导入语句的拼写，确保和文件名一致
2. 补全dynamic_dags.py的依赖导入和变量定义
3. 确保两个文件都放在Airflow的dags扫描目录下，如果存放在子目录，需要在子目录下添加空的__init__.py文件

修复后代码示例 #

dynamic_dags.py：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago

# 补全默认参数定义
args = {
    "owner": "airflow"
}

def generate_dag(name):
    with DAG(
        dag_id=f'dag_{name}',
        default_args=args,
        start_date=days_ago(2),
        schedule_interval='5 5 * * *',
        tags=['Test'],
        catchup=False
    ) as dag:
        dummy_task=DummyOperator(
            task_id="dynamic_dummy_task",
            dag=dag
        )
    return dag

load_dags.py：

# 修正导入的文件名，添加末尾的s
from dynamic_dags import generate_dag

globals()["Dynamic_DAG_A"] = generate_dag('A')

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