如何在Airflow中创建可多次运行的循环任务？

你需要的是通过循环创建多个独立的PythonOperator实例，并配置它们按顺序依次执行。直接重复赋值给同一个变量会覆盖之前的实例，无法构建有效依赖，正确的实现方式如下：

1. 定义任务执行函数 #

先写好每个PythonOperator要调用的业务函数：

def my_task_logic(task_index):
    print(f"正在执行第 {task_index + 1} 个任务")
    # 这里可以添加你的具体业务逻辑

2. 循环生成任务并配置依赖 #

通过列表存储每个任务实例，再依次设置依赖关系：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from airflow.utils.helpers import chain

with DAG(
    dag_id="sequential_loop_tasks",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    task_instances = []
    
    # 循环生成5个独立的PythonOperator
    for i in range(5):
        current_task = PythonOperator(
            task_id=f"step_task_{i}",  # 必须保证每个task_id唯一
            python_callable=my_task_logic,
            op_kwargs={"task_index": i}  # 传递参数区分不同任务
        )
        task_instances.append(current_task)
    
    # 方式一：手动循环配置依赖
    for idx in range(1, len(task_instances)):
        task_instances[idx-1] >> task_instances[idx]
    
    # 方式二：用Airflow内置的chain函数快速串联（二选一即可）
    # chain(*task_instances)

关键注意事项 #

• 每个任务的task_id必须唯一，这里通过循环变量i拼接字符串实现，避免Airflow报错
• 必须把每个生成的任务实例存入列表，否则后续循环会覆盖之前的实例，无法构建依赖链
• chain函数是Airflow提供的便捷工具，能快速将列表中的任务按顺序串联，和手动循环配置依赖效果完全一致

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