Airflow动态任务执行：如何限制最大并行数为4处理100条数据？

要实现单DAG Run内同时运行最多4个动态任务、完成一个自动调度下一个的需求，可通过Airflow内置参数实现，以下是具体方案：

1. 针对映射任务单独限制并发 #

使用任务的task_concurrency参数，该参数仅作用于当前任务的所有实例（包括动态映射生成的子任务），不影响DAG中其他任务的执行。

2. 限制整个DAG Run的任务并发总数 #

在DAG定义中设置max_active_tis_per_dag_run参数，控制单个DAG Run中同时运行的所有任务实例总数，适合整体控制DAG的资源占用。

示例代码 #

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

def fetch_id_list():
    # 模拟从数据源获取100条id的列表
    return list(range(100))

def process_single_id(id):
    # 替换为实际的id处理逻辑
    print(f"Processing id: {id}")

with DAG(
    dag_id="id_processing_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    # 可选：限制整个DAG Run的总并发任务数
    # max_active_tis_per_dag_run=4,
) as dag:

    fetch_ids_task = PythonOperator(
        task_id="fetch_id_list",
        python_callable=fetch_id_list,
    )

    process_ids_task = PythonOperator(
        task_id="process_single_id",
        python_callable=process_single_id,
        op_args=["{{ item }}"],
        # 限制该任务的最大并发实例数为4
        task_concurrency=4,
    ).expand(op_args=fetch_ids_task.output)

    fetch_ids_task >> process_ids_task

参数说明 #

• task_concurrency=4：仅对process_single_id任务的所有实例生效，保证最多同时运行4个id处理任务，完成一个后自动调度下一个待处理的id。
• max_active_tis_per_dag_run=4：如果启用，整个DAG Run中最多同时运行4个任务实例（包括fetch_id_list任务运行时的实例），适合需要严格控制整体资源的场景。

可根据实际资源情况选择对应方案，两种方式均能满足你的需求。

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