如何在Airflow中通过Params动态创建n个任务

问题核心 #

DAG解析阶段是静态的，无法直接获取运行时通过UI传入的params值来循环创建任务，必须借助Airflow的**动态任务映射（Dynamic Task Mapping）**功能，该功能在Airflow 2.3+版本支持。

解决方案代码 #

import pendulum
from airflow.decorators import dag, task

@dag(
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["forest_hpo"],
    params={"n": 3}  # UI触发时可修改此值
)
def dynamic_test():

    @task
    def get_task_range(**context):
        # 从运行上下文获取UI传入的n值
        n = context["params"]["n"]
        return list(range(n))

    @task
    def model_training_task(i):
        print(f"执行训练任务: {i}")

    # 先获取任务序号列表，再动态映射生成对应数量的任务
    task_ranges = get_task_range()
    model_training_task.expand(i=task_ranges)

dynamic_test()

工作原理 #

1. 获取运行时参数：get_task_range任务通过**context参数获取DAG运行时的上下文，从中提取UI传入的n值，生成包含0到n-1的序号列表。
2. 动态生成任务：调用model_training_task.expand(i=task_ranges)，Airflow会根据列表中的每个元素自动创建对应数量的独立任务实例，数量完全由UI指定的n决定。

注意事项 #

• 确保Airflow版本在2.3及以上，动态任务映射是该版本后引入的功能。
• 触发DAG时，在UI的"Trigger DAG w/ config"页面修改params中的n值，即可生成对应数量的任务。

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