Airflow2 函数执行后动态创建DAG的实现方案咨询

失效根因 #

当前实现无法生成DAG的核心原因是完全偏离Airflow的DAG加载逻辑：

• Airflow Scheduler仅会在启动阶段、或按配置周期触发的DAG文件解析阶段，扫描dags目录下的Python脚本，全量执行脚本顶层代码后，收集全局命名空间内所有DAG类对象纳入调度范围
• 你把动态生成DAG的逻辑放在了@once调度的普通DAG的PythonOperator任务里，这段代码实际是在Worker进程的任务运行阶段才执行，既不会触发Scheduler重新扫描dags目录，Worker进程内修改的globals()也不会同步到Scheduler进程的全局命名空间，自然无法生成被调度识别的独立DAG。

正确实现路径 #

1. 脚本顶层完成外部数据拉取
   不要把查询数据库拿DATA列表的逻辑放到任何运行时任务里，直接在DAG定义脚本的顶层代码位置执行查询，拿到需要的业务数据列表。
   基础结构示例：

   # 路径：dags/dynamic_biz_dag.py
   from airflow import DAG
   from airflow.operators.python import PythonOperator
   from datetime import datetime
   import psycopg2
   
   # 脚本被Scheduler解析时直接执行的拉数逻辑
   def get_biz_data():
       conn = psycopg2.connect("实际数据库连接串")
       cur = conn.cursor()
       cur.execute("select biz_id from your_biz_config_table")
       data_list = [row[0] for row in cur.fetchall()]
       cur.close()
       conn.close()
       return data_list
   
   DATA_LIST = get_biz_data()
   

2. 顶层循环生成DAG并直接注入全局命名空间
   不要通过运行时任务修改全局变量，直接在顶层循环遍历拿到的DATA列表，生成每个独立DAG对象后赋值到全局。
   核心逻辑示例：

   def create_standard_dag(dag_id, schedule, base_args, biz_config):
       # 通用标准化DAG构造逻辑
       with DAG(
           dag_id=dag_id,
           default_args=base_args,
           schedule_interval=schedule,
           start_date=datetime(2024, 1, 1),
           catchup=False
       ) as dag:
           # 定义DAG下的业务任务
           run_task = PythonOperator(
               task_id="simple_task",
               python_callable=lambda: print(f"执行业务配置: {biz_config}")
           )
       return dag
   
   # 全局默认参数
   base_dag_args = {"owner": "airflow", "retries": 1}
   
   # 循环生成DAG，直接挂到全局
   for index, biz_item in enumerate(DATA_LIST):
       target_dag_id = f"hello_world_{index}"
       globals()[target_dag_id] = create_standard_dag(
           dag_id=target_dag_id,
           schedule="0 * * * *", # 按实际业务调度规则替换
           base_args=base_dag_args,
           biz_config=biz_item
       )
   

3. 调整Scheduler适配配置
   在airflow.cfg中配置两个核心参数，适配动态DAG的刷新和性能要求：
   • 调整dag_dir_list_interval为可接受的DAG刷新周期（例如300秒），Scheduler会按这个周期重新扫描dags目录、执行脚本拉取最新的DATA列表，动态更新生成的DAG，无需手动重启Scheduler
   • 开启max_active_dag_processing_percentage控制单批次解析的DAG数量，避免大数量动态DAG导致的解析阻塞

常见踩坑规避 #

• 顶层的数据库拉取逻辑必须加超时和异常捕获：如果数据库连接超时或查询失败，会直接导致整个脚本解析报错，所有DAG都无法生成，建议拉取失败时返回空列表跳过生成，不要让脚本直接崩溃
• 循环生成DAG的代码不能被条件块遮挡：不要把逻辑放在if __name__ == "__main__"判断块内、或者嵌套在其他需要特定触发条件才执行的函数里，必须保证Scheduler解析脚本时代码直接执行
• 大数量动态DAG不要拆成单文件：单文件内循环生成DAG是Airflow官方认可的性能最优方案，比每个DAG单独写一个文件的IO开销低70%以上

内容的提问来源于stack exchange，提问作者Soumil Nitin Shah