Airflow v2.3.4：配置KubernetesPodOperator实现DAG任务并行及并发限制

问题核心 #

需要实现两个目标：

• 同一执行日期下的不同任务（task1_today、task2_today…taskN_today）全部并行运行，解决当前部分任务灰色「已调度」的问题
• 同一任务的不同日期实例（task1_today与task1_yesterday）不能同时运行，避免资源冲突

现有配置的问题 #

当前KubernetesPodOperator中设置的max_active_tis_per_dag=1是关键错误：这个参数限制整个DAG的所有任务实例最多只能有1个处于活跃状态，直接导致同DAG内的任务无法并行，大量任务被卡在调度队列。

修正配置步骤 #

1. 调整DAG级并行限制 #

将max_active_tasks明确配置在DAG定义中（而非仅放在default_args），值设为任务列表长度len(LIST_OF_TASKS)，确保同一执行日期下的所有任务能同时启动。

2. 设置任务级互斥限制 #

把原任务中的max_active_tis_per_dag=1替换为任务级并发限制参数max_active_tis_per_task=1（Airflow 2.2+版本适用），该参数会限制同一个task_id的不同执行日期实例只能有1个处于活跃状态，刚好满足同任务跨日期的互斥需求。

• 若使用Airflow 2.2之前的版本，替换为task_concurrency=1即可，功能完全等价。

3. 保留必要参数 #

default_args中的depends_on_past=False保持不变，确保任务不依赖上一周期的执行结果，不影响同日期任务的并行启动。

修改后的完整代码示例 #

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
import datetime
from datetime import timedelta
import kubernetes.client as k8s

LIST_OF_TASKS = ["task1", "task2", ..., "taskN"]  # 你的任务列表
DAG_IMAGE = "your-image:tag"
compute_resources = k8s.V1ResourceRequirements(
    requests={"cpu": "1", "memory": "2Gi"},
    limits={"cpu": "2", "memory": "4Gi"}
)

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email_on_failure": True,
    "email": ["intelligence@profinda.com"],
    "retries": 2,
    "retry_delay": timedelta(hours=6),
    "email_on_retry": False,
    "image_pull_policy": "Always",
}

# DAG定义时指定max_active_tasks，确保同日期任务全并行
with DAG(
    dag_id="your_dag_id",
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    start_date=datetime.datetime(2024, 1, 1),
    catchup=False,  # 不需要补历史任务可设为False，按需调整
    max_active_tasks=len(LIST_OF_TASKS),
) as dag:
    for spider_name in LIST_OF_TASKS:
        normalised_name = spider_name.lower().replace("_", "-")
        KubernetesPodOperator(
            namespace="airflow",
            service_account_name="airflow",
            image=DAG_IMAGE,
            image_pull_secrets=[k8s.V1LocalObjectReference("docker-registry")],
            container_resources=compute_resources,
            env_vars={
                "EXECUTION_DATE": "{{ execution_date }}",
            },
            cmds=["python3", "launcher.py", "-n", spider_name, "-r", "43000"],
            is_delete_operator_pod=True,
            in_cluster=True,
            name=f"Crawler-{normalised_name}",
            task_id=f"hydra-crawler-{normalised_name}",
            get_logs=True,
            max_active_tis_per_task=1,  # 同一task_id跨日期实例互斥
            # Airflow 2.2以下版本替换为：task_concurrency=1
        )

参数说明 #

• max_active_tasks：DAG级参数，控制同一执行日期下最多可同时运行的任务数，设为任务列表长度即可让同日期所有任务并行启动。
• max_active_tis_per_task：任务级参数，控制单个task_id的所有执行日期实例中，最多有1个处于活跃状态，实现task1_today与task1_yesterday的互斥。

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