Airflow中以Pythonic方式给DAG传列表参数并遍历执行任务的方法

问题分析 #

你遇到的两个核心问题原因如下：

1. 直接将列表传入DAG函数时，Airflow会把参数封装为DagParam对象（延迟求值的占位符），该对象无法直接迭代，因此报错DagParam object is not iterable。
2. 使用@dag装饰器的params参数时，参数定义仅作为schema校验，实际传入的值需要从dag_run.conf中获取，而非直接从**kwargs的顶层键值对读取。

最优实现方案 #

方案1：静态列表（DAG解析时确定） #

如果列表内容是固定的，直接在DAG定义中遍历生成任务即可，无需使用Airflow的参数封装：

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

@task
def process(item: str):
    # 处理逻辑示例
    print(f"Processing item: {item}")

@dag(
    dag_id="static_list_dag",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
)
def static_list_workflow():
    # 静态列表，可直接遍历生成任务
    items = ["a", "b", "c"]
    for item in items:
        process(item)

# 实例化DAG
static_list_workflow()

如果需要从外部配置（如Airflow变量）读取静态列表：

from airflow.models import Variable

@dag(
    dag_id="static_list_from_var_dag",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
)
def static_list_from_var_workflow():
    # 从Airflow变量读取列表（需提前在UI中设置变量`my_process_items`，值为JSON数组）
    items = Variable.get("my_process_items", deserialize_json=True)
    for item in items:
        process(item)

static_list_from_var_workflow()

方案2：动态列表（运行时传入，推荐） #

如果需要在触发DAG时动态传入列表，使用Dynamic Task Mapping（Airflow 2.3+支持），这是最符合Python风格的动态任务生成方式：

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

@task
def process(item: str):
    print(f"Processing item: {item}")

@task
def fetch_dynamic_items(**kwargs):
    # 从触发DAG时传入的conf中获取列表，默认值为空列表
    return kwargs["dag_run"].conf.get("items", [])

@dag(
    dag_id="dynamic_mapping_dag",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False
)
def dynamic_list_workflow():
    # 获取动态传入的列表
    items = fetch_dynamic_items()
    # 使用expand方法自动为每个列表元素生成子任务
    process.expand(item=items)

dynamic_list_workflow()

触发方式

在Airflow UI中触发DAG时，在Configuration中输入JSON格式的参数：

{"items": ["a", "b", "c"]}

方案3：使用params参数做schema校验 #

如果需要对传入的列表做格式校验，可结合@dag的params参数，同时从dag_run.conf中读取实际值：

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

@task
def process(item: str):
    print(f"Processing item: {item}")

@task
def get_validated_items(**kwargs):
    # 优先取触发时传入的conf，若无则用params中定义的默认值
    items = kwargs["dag_run"].conf.get("items", kwargs["params"]["items"])
    return items

@dag(
    dag_id="validated_list_dag",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
    # 定义参数schema，限制为数组类型，默认值为["a", "b", "c"]
    params={"items": Param(["a", "b", "c"], type="array", description="Items to process")}
)
def validated_list_workflow():
    items = get_validated_items()
    process.expand(item=items)

validated_list_workflow()

关键注意点 #

• Airflow的任务是在DAG解析阶段生成的，因此直接在DAG函数中迭代DagParam对象会失败（此时参数未被实际赋值）。
• Dynamic Task Mapping是Airflow官方推荐的动态任务生成方式，无需手动循环，语法简洁且符合Python风格。

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