Apache Airflow：能否配置每个Worker的DAG-Runs并发数量？

Airflow原生没有直接配置每个Worker上DAG-Run数量的参数，因为它的调度逻辑核心是任务级别的分配，而非将整个DAG-Run绑定到特定Worker。针对你遇到的场景，下面给出具体的解决方案和最优实践：

你的场景问题分析 #

你当前的配置中，worker-concurrency=2允许每个Worker同时运行2个任务，但Scheduler不会主动保证同一个DAG-Run的两个任务（A和B）分配到同一个Worker，因此会出现单个Worker跑两个A或两个B的情况——这是Airflow默认调度逻辑的正常表现。

可行的解决方案 #

1. 利用任务亲和性规则（KubernetesExecutor适用） #

如果你的Airflow部署在Kubernetes环境下，可以通过Pod亲和性配置，强制将同一个DAG-Run的所有任务调度到同一个Worker节点。具体可以在任务的Operator中添加pod_override参数，基于dag_run_id标识设置亲和性规则：

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from kubernetes.client import models as k8s

task_a = KubernetesPodOperator(
    task_id="task_a",
    name="task-a",
    namespace="airflow",
    image="your-image:latest",
    pod_override=k8s.V1Pod(
        spec=k8s.V1PodSpec(
            affinity=k8s.V1Affinity(
                pod_affinity=k8s.V1PodAffinity(
                    required_during_scheduling_ignored_during_execution=[
                        k8s.V1PodAffinityTerm(
                            label_selector=k8s.V1LabelSelector(
                                match_expressions=[
                                    k8s.V1LabelSelectorRequirement(
                                        key="dag_run_id",
                                        operator="In",
                                        values=["{{ dag_run.id }}"]
                                    )
                                ]
                            ),
                            topology_key="kubernetes.io/hostname"
                        )
                    ]
                )
            )
        )
    ),
    # 其他任务参数...
)

这样同一个DAG-Run的任务A和B会被调度到同一Worker节点，确保每个Worker最多处理一个DAG-Run的两个任务。

2. 专用Worker组+队列绑定（CeleryExecutor适用） #

如果使用CeleryExecutor，可以为My-DAG创建专用的Worker组：

• 启动12个Worker，指定专属队列（比如my_dag_queue），并设置每个Worker的worker-concurrency=2
• 在My-DAG的定义中添加queue='my_dag_queue'，让该DAG的所有任务只进入这个专属队列
• 同时保持max_active_runs_per_dag=12，这样每个Worker会分配到一个DAG-Run的两个任务，避免同类型任务扎堆的情况

3. 基于任务资源的精准调度 #

重新梳理任务的资源消耗（CPU、内存），为每个任务设置明确的资源请求和限制（K8s环境下），或者在CeleryWorker中配置资源阈值。Airflow的Scheduler会根据Worker的剩余资源分配任务，减少单个Worker过载或任务分配不均衡的情况。

最优实践总结 #

• 如果是K8s部署，优先使用Pod亲和性实现DAG-Run与Worker的绑定，这是最精准的方案
• 如果是Celery部署，用专用队列+Worker组隔离特定DAG的任务，简化调度逻辑
• 无论哪种部署方式，明确任务的资源需求都是基础，能让调度更合理，减少不必要的问题

内容的提问来源于stack exchange，提问作者Johannes-R-Schmid