动态生成 DAG

本文讨论动态生成结构、但各次 DAG Run 之间任务数量不变的 DAG。如果希望任务数量根据上游输出或结果改变，请参阅“动态任务映射”；从 Airflow 2.6 起，任务组也支持这类映射。

注意

保持任务与任务组的生成顺序一致

动态生成 DAG 时，每次都应按相同顺序生成任务和任务组，否则每次刷新 Grid View 都可能看到它们重新排序。可以在数据库查询中使用稳定排序，也可以在 Python 中调用 sorted()。

通过环境变量生成动态 DAG

如果需要通过变量配置代码，顶层代码应使用环境变量。在顶层读取 Airflow Variables 会为了取值而连接元数据库，拖慢解析并增加数据库负载。有关在 DAG 中通过 Jinja 模板合理使用 Airflow Variables 的方法，参阅相关最佳实践。

例如，在生产环境将 DEPLOYMENT 设置为 PROD，在开发环境设置为 DEV，再按环境变量值生成不同的 DAG：

deployment = os.environ.get("DEPLOYMENT", "PROD")
if deployment == "PROD":
    task = Operator(param="prod-param")
elif deployment == "DEV":
    task = Operator(param="dev-param")

生成内嵌元数据的 Python 代码

可以在外部生成 Python 代码，将元数据保存为可导入的常量。DAG 直接导入这些常量，用它们构建对象和依赖关系。多个 DAG 可以共享这些代码，无需各自查找、加载和解析元数据；Python 解释器处理 import 时会完成相关工作。这种方法听起来有些特殊，但生成合法、可导入的 Python 代码其实很简单。

例如，在 DAG 目录中动态生成 my_company_utils/common.py：

# This file is generated automatically !
ALL_TASKS = ["task1", "task2", "task3"]

然后在各个 DAG 中导入并使用 ALL_TASKS：

from my_company_utils.common import ALL_TASKS

with DAG(
    dag_id="my_dag",
    schedule=None,
    start_date=datetime(2021, 1, 1),
    catchup=False,
):
    for task in ALL_TASKS:
        # create your operators and relations here
        ...

别忘了在 my_company_utils 中添加空的 __init__.py，并在使用默认 glob 语法的 .airflowignore 中加入 my_company_utils/*，让调度器查找 DAG 时忽略整个工具目录。

通过结构化数据文件配置动态 DAG

如果元数据更复杂，且希望用非 Python 的结构化格式保存，应将它导出成文件并随 DAG 推送到 DAG 目录，而不是在 DAG 顶层代码中拉取。原因见“顶层 Python 代码”部分。

JSON、YAML 都是合适的格式。最好将元数据与加载它的 DAG 模块放在同一包或目录中，这样容易定位文件。可以利用包含 DAG 的模块的 __file__ 属性：

my_dir = os.path.dirname(os.path.abspath(__file__))
configuration_file_path = os.path.join(my_dir, "config.yaml")
with open(configuration_file_path) as yaml_file:
    configuration = yaml.safe_load(yaml_file)
# Configuration dict is available here

注册动态 DAG

使用 @dag 装饰器或 with DAG(..) 上下文管理器动态生成 DAG 时，Airflow 会自动注册它们。

from datetime import datetime
from airflow.sdk import dag, task

configs = {
    "config1": {"message": "first Dag will receive this message"},
    "config2": {"message": "second Dag will receive this message"},
}

for config_name, config in configs.items():
    dag_id = f"dynamic_generated_dag_{config_name}"

    @dag(dag_id=dag_id, start_date=datetime(2022, 2, 1))
    def dynamic_generated_dag():
        @task
        def print_message(message):
            print(message)

        print_message(config["message"])

    dynamic_generated_dag()

以上代码为每组配置生成一个 DAG：dynamic_generated_dag_config1 与 dynamic_generated_dag_config2，两者可分别使用相应配置运行。

如果不希望自动注册，可以为 DAG 设置 auto_register=False。

2.4 版本变更：从 2.4 起，调用 @dag 装饰的函数创建的 DAG，以及在 with DAG(...) 中使用的 DAG，都会自动注册，无需再存入全局变量。

减少执行期间的 DAG 解析延迟

此功能自 2.4 起提供。

如果一个 DAG 文件生成了很多动态 DAG，任务执行前解析该文件可能产生不必要的延迟，表现为任务启动前的等待。

原因是：在任务执行之前，Airflow 会再次解析定义该 DAG 的 Python 文件。

DAG File Processor 需要加载整个文件来处理所有元数据，但任务执行只需要一个 DAG 对象。因此，可以在执行任务时跳过不需要的 DAG 对象生成，缩短解析时间；生成的 DAG 越多，收益通常越明显。

这种优化并非总能使用。如果后续 DAG 的生成依赖先前 DAG，或生成过程带有副作用，就需要格外谨慎并充分测试。

原文链接的“Airflow’s Magic Loop”文章展示过将任务执行阶段的解析耗时从 120 秒降至 200 毫秒的例子。该文章写于 Airflow 2.4 之前，使用的是未文档化行为，而非下文的 get_parsing_context()。

在 Airflow 2.4 及以后版本，可以使用有正式文档的 get_parsing_context() 获取当前上下文。

遍历生成 DAG 的集合时，可以根据上下文决定生成全部 DAG（DAG File Processor 解析时），还是只生成一个 DAG（执行任务时）。

get_parsing_context() 返回 AirflowParsingContext。如果只需要一个 DAG／任务，其中的 dag_id 和 task_id 会被设置；需要完整解析时（如 DAG File Processor），这两个字段均为 None。

from airflow.sdk import DAG
from airflow.sdk import get_parsing_context

current_dag_id = get_parsing_context().dag_id

for thing in list_of_things:
    dag_id = f"generated_dag_{thing}"
    if current_dag_id is not None and current_dag_id != dag_id:
        continue  # skip generation of non-selected Dag

    with DAG(dag_id=dag_id, ...):
        ...