Airflow v2如何基于UI输入的数组参数创建动态任务映射？

问题分析 #

你当前的写法将模板字符串用单引号包裹，导致Airflow未解析Jinja表达式，而是把{{ dag_run.conf.param1 }}当成普通字符串处理，进而遍历字符串的每个字符生成任务，不符合需求。

解决方案 #

去掉模板表达式的引号包裹，让Airflow正确解析Jinja模板，将其渲染为数组类型，这样.expand()就能根据数组元素数量生成对应任务。

正确代码示例 #

from airflow.decorators import dag, task
from airflow.models.param import Param
from datetime import datetime

@dag(
    start_date=datetime(2021, 1, 1),
    schedule=None,
    params={
        "param1": Param(
            ["all"],
            "Select from the list of options.",
            type="array",
            title="Param 1",
            examples=["all", 'half'],
        )
    }
)
def updateDAG():
    @task
    def task1(arg1):
        # 示例任务逻辑
        print(f"执行任务，参数: {arg1}")
        return arg1

    # 正确的动态任务映射写法
    task1.override(task_id="run_task").expand(arg1="{{ params.param1 }}")

updateDAG()

补充说明 #

• 使用{{ params.param1 }}可直接引用DAG定义的参数，同时兼容手动触发时传入的自定义数组参数（会自动覆盖默认值）。
• 如果需要明确从触发配置中获取参数，也可写成{{ dag_run.conf.param1 }}，建议加上兜底逻辑避免触发时未传参的情况：{{ dag_run.conf.get('param1', params.param1) }}。
• 由于你在Param中定义了type="array"，Airflow UI会强制用户输入数组类型的参数，确保.expand()能获取到合法的可迭代对象。

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