Airflow KubernetesExecutor: запуск каждой задачи в отдельном поде

Запуск каждой задачи Airflow в отдельном поде через KubernetesExecutor — это переход от «кошек в одной коробке» к строгой изоляции процессов. Если вы устали от конфликтов зависимостей между DAG’ами или вам нужна гибкость в выборе образа и ресурсов под конкретную задачу, то это ваш путь. Гайд для тех, кто уже пережил боль shared-окружения и хочет сделать по-уму.

Базовая конфигурация executor

В airflow.cfg или через переменные окружения (AIFLOW__KUBERNETESES__…) задаём ключевые параметры. Основное — указать executor и настроить подключение к кластеру. In-cluster конфигурация обычно работает сама, если под Airflow запущен в Kubernetes.

[core]
executor = KubernetesExecutor

[kubernetes_executor]
namespace = airflow-tasks
pod_template_file = /opt/airflow/pod_templates/pod_template.yaml
worker_container_repository = apache/airflow
worker_container_tag = 2.7.3
delete_worker_pods = true
delete_worker_pods_on_failure = true

Сервисный аккаунт и RBAC

Поду задачи будут запускаться под определённым ServiceAccount. Ему нужны права на создание/удаление подов в целевом namespace. Создаём аккаунт и минимальную роль:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: airflow-task-runner
  namespace: airflow-tasks
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: airflow-tasks
  name: pod-creator
rules:
- apiGroups: [""]
  resources: ["pods"]
  verbs: ["create", "get", "list", "watch", "delete"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: airflow-task-runner-binding
  namespace: airflow-tasks
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: pod-creator
subjects:
- kind: ServiceAccount
  name: airflow-task-runner
  namespace: airflow-tasks

Укажите этот аккаунт в pod_template_file.

Сердце системы: pod_template_file

Это YAML-манифест, шаблон для всех рабочих подов. Определите здесь базовый образ, serviceAccountName, volumeMounts и переменные окружения, общие для всех задач. Критически важно задать AIRFLOW__CORE__EXECUTOR=KubernetesExecutor, иначе под попытается использовать LocalExecutor.

apiVersion: v1
kind: Pod
metadata:
  labels:
    app: airflow-task
spec:
  serviceAccountName: airflow-task-runner
  restartPolicy: Never
  containers:
    - name: base
      image: apache/airflow:2.7.3
      imagePullPolicy: IfNotPresent
      env:
        - name: AIRFLOW__CORE__EXECUTOR
          value: KubernetesExecutor
        - name: AIRFLOW__DATABASE__SQL_ALCHEMY_CONN
          valueFrom:
            secretKeyRef:
              name: airflow-metadata
              key: connection
      resources:
        requests:
          memory: "256Mi"
          cpu: "100m"

Динамическое управление ресурсами через executor_config

Шаблон — это основа, но конкретную задачу можно донастроить прямо в DAG. Через параметр executor_config в операторе задаются требования к CPU, память, или даже целый другой образ.

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

def process_data():
    import pandas as pd
    print("Processing with pandas")

with DAG('k8s_example', start_date=datetime(2023, 1, 1)) as dag:
    heavy_task = PythonOperator(
        task_id='heavy_data_processing',
        python_callable=process_data,
        executor_config={
            "pod_override": {
                "spec": {
                    "containers": [{
                        "name": "base",
                        "resources": {
                            "requests": {"memory": "2Gi", "cpu": "1000m"},
                            "limits": {"memory": "4Gi", "cpu": "2000m"}
                        }
                    }]
                }
            }
        }
    )

Когда под не запускается: чек-лист для отладки

1. Проверьте логи scheduler. Там будут ошибки, если не удалось создать под. Частая причина — неверное имя service account или недостаток прав RBAC.
2. Убедитесь, что pod_template_file валиден. Проверьте через kubectl apply --dry-run=client -f pod_template.yaml.
3. Образ. Указанный в шаблоне образ должен существовать и быть доступен из кластера. Ошибка ImagePullBackOff — явный признак.
4. Контекст запуска. Scheduler должен работать внутри кластера и иметь доступ к Kubernetes API. Для внешнего кластера нужно настроить kube_config и in_cluster=False.
5. Лимиты ресурсов. Если запрашиваете больше ресурсов, чем доступно в namespace, под будет в статусе Pending. Смотрите kubectl describe pod.

Подводные камни

Главный — увеличение нагрузки на etcd и API Kubernetes при высокой частоте запуска коротких задач (тысячи подов в час). Планируйте ресурсы нод с запасом на быструю диспетчеризацию. Второй — наследование конфигурации. Переменные из pod_template_file и executor_config мержатся, при конфликте побеждает executor_config. Всегда тестируйте итоговый манифест.

KubernetesExecutor даёт чистую изоляцию и невероятную гибкость, ценой некоторой сложности отладки. Потратьте время на настройку шаблона и RBAC один раз, и вы получите мощный и предсказуемый оркестратор задач.