如何在MWAA/Airflow中基于任务列表创建动态顺序任务（非动态任务映射）

核心逻辑 #

基于任务列表生成顺序任务链的核心是遍历任务列表，逐个创建任务实例并依次设置依赖关系，最终将start、动态任务链、end串联为完整工作流。

具体实现步骤 #

• 定义基础任务模板：先确定start、end任务，以及动态任务的通用执行逻辑（可根据业务需求选择PythonOperator、BashOperator等）。
• 初始化任务链起点：将start任务作为任务链的初始上游节点。
• 遍历生成动态任务：循环处理任务列表中的每个任务名称，创建对应的任务实例，将其与前一个任务设置依赖（>>），并更新当前上游节点为新创建的任务。
• 收尾连接end任务：遍历完成后，将最后一个动态任务与end任务设置依赖，完成整个任务链构建。

代码示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# 动态任务的通用执行函数
def execute_task(task_name):
    print(f"Running task: {task_name}")
    # 替换为实际业务逻辑

with DAG(
    dag_id="dynamic_sequential_workflow",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False,
    tags=["dynamic_tasks"]
) as dag:
    # 定义start和end任务
    start = PythonOperator(
        task_id="start",
        python_callable=lambda: print("Workflow started")
    )

    end = PythonOperator(
        task_id="end",
        python_callable=lambda: print("Workflow finished")
    )

    # 自定义动态任务列表
    task_list = ["task1", "task2", "task3"]

    # 初始化当前上游任务为start
    current_task = start

    # 循环创建动态任务并串联
    for task_id in task_list:
        dynamic_task = PythonOperator(
            task_id=task_id,
            python_callable=execute_task,
            op_kwargs={"task_name": task_id}
        )
        current_task >> dynamic_task
        current_task = dynamic_task

    # 连接最后一个动态任务到end
    current_task >> end

注意事项 #

• 任务列表中的task_id必须唯一，避免重复导致DAG初始化失败。
• 若需要不同类型的动态任务（如部分用BashOperator），可通过字典映射任务名称与对应Operator类型，在循环中动态选择。
• MWAA环境下的实现逻辑与本地Airflow一致，只需确保DAG代码上传至MWAA指定的S3路径，且依赖包已在MWAA环境配置。

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