Apache Airflow动态生成DAG自动消失问题咨询

这绝对不是Airflow动态DAG的预期行为，问题出在你创建动态DAG的逻辑上——你把DAG的创建和主DAG的运行绑定得太紧密了，忽略了Airflow加载DAG的核心机制。

为什么动态DAG会消失？ #

Airflow的调度器会定期（默认每30秒）扫描你的DAG文件，重新执行文件里的代码，然后根据globals()里存在的DAG对象来维护UI和可用DAG列表。

你原来的代码只有在主DAG运行、并且获取到对应数据库条目时，才会把那个动态DAG的对象放到globals()里。当主DAG下次运行没拿到旧条目时，DAG文件重新被调度器扫描时，那段创建旧DAG的代码不会执行，globals()里就没有这个DAG的对象了——调度器会认为这个DAG已经被删除，自然从UI和list_dags里移除它。

正确的动态DAG创建方式 #

你需要把动态DAG的生成逻辑和主DAG的运行解耦：让DAG文件在每次调度器扫描时，都主动从数据库中获取所有需要存在的DAG条目（包括已经创建好的、Dag_id不为空的条目），然后为每个条目生成对应的DAG对象并放入globals()。主DAG只负责更新数据库的状态（比如给新条目生成Dag_id），不直接创建DAG。

举个具体的代码示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
import datetime
import psycopg2  # 假设你用PostgreSQL，根据你的数据库类型调整

# 从数据库获取所有需要生成DAG的条目（包括已存在的）
def get_all_dag_entries():
    conn = psycopg2.connect("dbname=your_db user=your_user password=your_pass host=your_host")
    cur = conn.cursor()
    # 查询所有Dag_id不为空的条目（即已经需要存在的动态DAG）
    cur.execute("SELECT id, dag_id, schedule FROM your_table WHERE dag_id IS NOT NULL")
    entries = cur.fetchall()
    cur.close()
    conn.close()
    return entries

# 定义默认参数
default_args = {
    'owner': 'airflow',
    'start_date': datetime.datetime(2018, 1, 1),
    'catchup': False  # 避免自动调度历史任务，按需调整
}

# 循环生成所有需要的DAG
for entry in get_all_dag_entries():
    entry_id, dag_id, schedule = entry

    # 封装DAG创建逻辑
    def create_dag(dag_id, schedule, entry_id, default_args):
        dag = DAG(
            dag_id=dag_id,
            default_args=default_args,
            schedule_interval=schedule
        )
        with dag:
            # 这里定义你的任务逻辑，比如一个简单的打印任务
            def run_task():
                print(f"Executing task for entry {entry_id} in DAG {dag_id}")
            
            PythonOperator(
                task_id='run_main_task',
                python_callable=run_task
            )
        return dag

    # 将DAG对象放入globals()，让调度器能识别到
    globals()[dag_id] = create_dag(dag_id, schedule, entry_id, default_args)

主DAG的职责调整 #

你的主DAG只需要完成以下工作：

1. 查询数据库中Dag_id为空的新条目
2. 为每个新条目生成唯一的dag_id（比如dagid-{entry_id}）
3. 更新数据库中该条目的dag_id字段

这样，当主DAG运行更新完数据库后，下一次调度器扫描DAG文件时，就会自动加载这个新的动态DAG；而已经存在的DAG，因为数据库里有对应的条目，每次扫描都会被重新生成，不会消失。

额外注意事项 #

• 确保get_all_dag_entries()能稳定获取所有需要存在的DAG条目，避免因为数据库查询问题导致DAG漏加载。
• 动态DAG的dag_id必须唯一且固定，不能每次生成都变化，否则会被识别为新的DAG。
• 如果你的DAG有依赖或者复杂逻辑，建议把DAG创建的封装函数写得更健壮，比如添加异常处理、日志输出等。

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