使用KubernetesExecutor的Airflow仅单Pod运行，如何开启多Pod并行？

问题场景 #

使用指定Chart和values.yaml部署Airflow后，运行包含20个并行任务的DAG时，仅单个任务Pod启动，所有任务无法并行执行：

部署配置片段 #

Chart.yaml：

apiVersion: v2
name: airflow
description: Umbrella chart for Airflow
type: application
version: 0.0.1
appVersion: "2.1.2"
dependencies:
  - name: airflow
    alias: airflow
    version: 8.9.0
    repository: https://airflow-helm.github.io/charts

values.yaml核心配置：

airflow:
  airflow:
    executor: KubernetesExecutor
    legacyCommands: false
    image:
      repository: apache/airflow
      tag: 2.8.4-python3.9
    config:
      AIRFLOW__WEBSERVER__EXPOSE_CONFIG: "True"
      AIRFLOW__CORE__LOAD_EXAMPLES: "True"
  workers:
    enabled: false

DAG代码 #

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

def heavy_computation(task_number):
    sleep_time = random.uniform(0, 1)
    time.sleep(sleep_time)
    print(f"Task {task_number} completed after {sleep_time:.2f} seconds")

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 3, 21),
    'retries': 0,
}

with DAG(
    'parallel_computation_dag',
    default_args=default_args,
    schedule_interval=None,
    catchup=False,
    max_active_tasks=10,
) as dag:
    tasks = [
        PythonOperator(
            task_id=f'heavy_task_{i}',
            python_callable=heavy_computation,
            op_kwargs={'task_number': i},
        ) for i in range(20)
    ]

问题原因 #

虽然指定了KubernetesExecutor，但缺少全局并行度、调度线程数等关键配置，Airflow调度能力被限制，无法同时启动多个任务Pod。

解决步骤 #

修改values.yaml，补充并行执行相关配置：

1. 补充Airflow核心并行参数 #

在airflow.airflow.config节点下添加以下配置：

airflow:
  airflow:
    config:
      # 全局允许的最大并行任务数，根据集群资源调整
      AIRFLOW__CORE__PARALLELISM: "20"
      # 单个DAG允许的最大并行任务数，与DAG的max_active_tasks匹配
      AIRFLOW__CORE__DAG_CONCURRENCY: "10"
      # Scheduler线程数，决定同时调度任务的能力
      AIRFLOW__SCHEDULER__MAX_THREADS: "10"
      # KubernetesExecutor专属配置：每个任务Pod仅运行1个任务
      AIRFLOW__KUBERNETES__WORKER_CONCURRENCY: "1"
      # 任务完成后自动清理Pod
      AIRFLOW__KUBERNETES__DELETE_WORKER_PODS: "True"

2. 优化Scheduler资源配置 #

给Scheduler分配足够资源，避免因资源不足导致调度能力受限：

airflow:
  scheduler:
    resources:
      requests:
        cpu: "1"
        memory: "2Gi"
      limits:
        cpu: "2"
        memory: "4Gi"

3. 验证ServiceAccount权限 #

KubernetesExecutor需要Airflow的ServiceAccount具备创建、删除Pod的权限。默认情况下airflow-helm chart会自动配置RBAC规则，若未生效，可手动绑定权限：

kubectl create rolebinding airflow-edit --clusterrole=edit --serviceaccount=<your-namespace>:airflow -n <your-namespace>

4. 更新部署 #

应用修改后的配置：

helm upgrade airflow . -f values.yaml

验证 #

重新触发DAG后，通过以下方式确认并行效果：

• 执行kubectl get pods -n <your-namespace>，可看到多个任务Pod同时运行
• 在Airflow UI的任务列表中，多个任务处于Running状态

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