Запускалась ли каждая таска в airflow в отдельном pod-е

Задачи в Airflow и Pod-ы: как работает распределение

Краткий ответ

НЕ все таски в Airflow автоматически запускаются в отдельных pod-ах. Это зависит от executor (исполнителя), который используется. При использовании KubernetesExecutor или CeleryKubernetesExecutor каждая таска может быть запущена в отдельном pod-е, но это поведение настраивается.

Executors в Airflow

Airflow поддерживает несколько типов executors, и поведение запуска тасок зависит от выбора:

1. SequentialExecutor (для разработки)

• Запускает таски последовательно на одной машине
• Не использует pod-ы
• Только для разработки и тестирования

2. LocalExecutor (для малых нагрузок)

• Запускает таски параллельно на одной машине в разных процессах
• Не использует pod-ы
• Хороший для небольших DAG-ов

from airflow.executors.local_executor import LocalExecutor

# airflow.cfg
executor = LocalExecutor
max_parallel_tasks = 4  # Максимум 4 таски одновременно

3. CeleryExecutor (распределённое выполнение)

• Использует Celery (очередь сообщений) для распределения тасок
• Каждая таска выполняется на отдельном Celery worker узле
• Worker-ы работают как отдельные процессы, НЕ в pod-ах (если Celery на ВМ)
• Если Celery работает на Kubernetes, то может быть в pod-ах

# airflow.cfg
executor = CeleryExecutor
broker_url = redis://redis:6379/0
result_backend = db+postgresql://...

4. KubernetesExecutor (облачный подход)

• Каждая таска запускается в отдельном pod-е Kubernetes
• Это основной executor для production на K8s
• Максимальная изоляция и гибкость
• Накладные расходы: время на creation pod-а (несколько секунд)

# airflow.cfg
executor = KubernetesExecutor
kubernetes_cluster_name = my-cluster
kubernetes_namespace = airflow

5. CeleryKubernetesExecutor (гибридный)

• Комбинация Celery + Kubernetes
• Может запускать таски как в pod-ах, так и на Celery worker-ах
• Гибче всего, но сложнее в настройке

KubernetesExecutor: как работает

При использовании KubernetesExecutor, каждая таска оборачивается в отдельный pod:

Airflow Scheduler
    ↓
   DAG запущен
    ↓
   Scheduler создаёт pod для каждой таски
    ↓
  Pod #1 (таска 1) → выполняет код → удаляется
  Pod #2 (таска 2) → выполняет код → удаляется
  Pod #3 (таска 3) → выполняет код → удаляется

Пример DAG с KubernetesExecutor:

from airflow import DAG
from airflow.providers.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "data-team",
    "start_date": datetime(2024, 1, 1),
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG("k8s_example", default_args=default_args) as dag:
    
    # Явно используем KubernetesPodOperator
    task1 = KubernetesPodOperator(
        task_id="extract_data",
        image="my-docker-image:latest",
        cmds=["python", "extract.py"],
        namespace="airflow",
        get_logs=True,
    )
    
    task2 = KubernetesPodOperator(
        task_id="transform_data",
        image="my-docker-image:latest",
        cmds=["python", "transform.py"],
        namespace="airflow",
    )
    
    task1 >> task2

Настройка Airflow на Kubernetes

Как включить KubernetesExecutor:

# helm values.yaml для Airflow на K8s
executor: "KubernetesExecutor"

kubernetes:
  namespace: "airflow"
  affinity: {}
  tolerations: []
  
workers:
  replicas: 1
  resources:
    limits:
      cpu: "1000m"
      memory: "2Gi"

PodTemplate для кастомизации pod-ов:

from airflow.kubernetes.pod_generator import PodGenerator

pod_template_yaml = """
apiVersion: v1
kind: Pod
metadata:
  name: task-pod
spec:
  restartPolicy: Never
  containers:
  - name: task-container
    image: my-docker-image:latest
    resources:
      limits:
        memory: "1Gi"
        cpu: "500m"
"""

KubernetesPodOperator(
    task_id="my_task",
    pod_template_file="pod_template.yaml",
)

Сравнение executors

Преимущества и недостатки pod-ов для тасок

Преимущества:

• Полная изоляция тасок
• Независимые ресурсы (CPU, RAM) для каждой таски
• Автоматический cleanup после завершения
• Легко масштабировать на облаке
• Разные контейнеры для разных тасок

Недостатки:

• Overhead на создание pod-а (2-5 секунд)
• Дороже в облаке (каждый pod = вычислительные ресурсы)
• Медленнее чем процессы на одной машине
• Требует правильной настройки Kubernetes и security policies

Практические рекомендации

1. Используй KubernetesExecutor если:

   • Работаешь на облаке (AWS EKS, GCP GKE, Azure AKS)
   • Нужна максимальная масштабируемость
   • Таски требуют разных окружений или версий библиотек

2. Используй CeleryExecutor если:

   • Работаешь на своём датацентре
   • Нужна максимальная производительность
   • Есть опыт с Celery

3. Используй LocalExecutor если:

   • Разрабатываешь DAG локально
   • Нагрузка мала (< 100 тасок в день)

Выводы

Каждая таска в Airflow НЕ автоматически запускается в pod-е. Это зависит от выбранного executor-а. KubernetesExecutor — это выбор для облачных сценариев, где каждая таска действительно запускается в отдельном pod-е Kubernetes с полной изоляцией и гибкостью масштабирования.