Airflow 2.2.3中DAG无法加载，动态并行任务配置遇阻求助

问题 #

在Airflow 2.2.3中创建了一个包含3个任务的DAG，期望实现任务流：load() → make_jobs() → 多个do_job任务并行执行。编写了循环遍历job_list生成do_job任务的代码后，DAG无法在Airflow UI中显示，执行python my_dag_file.py命令也无法结束。将my_dag改为普通Python函数运行逻辑正常；移除循环代码则可正常编译。

原代码如下：

from datetime import datetime
from typing import List

from airflow.decorators import dag, task


@dag(dag_id='myDag',
     start_date=datetime(2023, 5, 15),
     schedule_interval="0 5 * * *",
     max_active_runs=1,
     catchup=False)
def my_dag():
    
    @task
    def load() -> List[dict]:
        ... # load some data from database

    @task
    def make_jobs(params: list) -> List[List[dict]]:
        ... # with params from database, make some list of jobs. One job is List[dict].

    @task
    def do_job(job: list) -> None:
        ... # do job

    param_list = load()
    job_list = make_jobs(param_list)

    for job in job_list:
        do_job(job)

dag = my_dag()

期望任务流：

load() --> make_jobs() --> job_1
                       |-> job_2
                       |-> job_3
                       |   ...
                       --> job_n

问题分析 #

问题出在for job in job_list这部分代码。Airflow的TaskFlow API中，job_list是一个TaskInstance对象，而非实际运行时的列表数据。在DAG解析阶段（执行python my_dag_file.py或Airflow UI加载DAG时），Airflow会尝试迭代这个TaskInstance，而非等到任务实际运行时才处理，这直接导致程序陷入无限循环或无法完成解析。

Airflow在解析DAG时需要确定所有任务的结构和依赖关系，而这里的循环依赖make_jobs任务的输出——这部分输出只有在任务运行时才会生成，解析阶段无法获取实际列表内容，因此直接迭代会引发解析逻辑异常。

解决方案 #

使用Airflow 2.2+支持的expand方法实现动态任务映射，替代手动循环。expand方法可根据上游任务的输出自动生成多个并行的do_job任务，且能被Airflow正确解析。

修改后的代码如下：

from datetime import datetime
from typing import List

from airflow.decorators import dag, task


@dag(dag_id='myDag',
     start_date=datetime(2023, 5, 15),
     schedule_interval="0 5 * * *",
     max_active_runs=1,
     catchup=False)
def my_dag():
    
    @task
    def load() -> List[dict]:
        ... # load some data from database

    @task
    def make_jobs(params: list) -> List[List[dict]]:
        ... # with params from database, make some list of jobs. One job is List[dict].

    @task
    def do_job(job: list) -> None:
        ... # do job

    param_list = load()
    job_list = make_jobs(param_list)
    
    # 使用expand方法动态生成并行任务
    do_job.expand(job=job_list)

dag = my_dag()

说明 #

• expand方法会告知Airflow，do_job任务需根据job_list的输出内容动态创建多个实例，每个实例对应job_list中的一个元素。
• 这种方式符合Airflow的DAG解析逻辑，解析阶段Airflow能识别这是动态任务映射，不会尝试直接迭代任务对象，从而避免无限循环或解析失败问题。
• 最终生成的任务流与预期完全一致：load() → make_jobs() → 多个并行的do_job任务。

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