Глоссарий — Apache Airflow 2

Класс систем, управляющих графом зависимых задач с моделью состояния, retry-логикой и единой наблюдаемостью. Airflow, Prefect, Dagster, Argo Workflows, Temporal — workflow engines. Отличие от scheduler (cron) — workflow engine знает связи между задачами и их состояния.

# Cron знает: 'запусти script.sh в 03:00 каждый день'
# Airflow знает: 'extract → transform → [load, notify]'
# и если extract падает, transform/load/notify получают upstream_failed

Направленный ациклический граф — основная абстракция Airflow. Узлы — это tasks, рёбра — зависимости (A → B значит 'B зависит от A'). Ациклический — нельзя пройти из вершины в неё саму. Гарантирует конечное число шагов выполнения.

@dag(schedule='@daily', start_date=datetime(2026, 1, 1))
def my_dag():
    extract = extract_task()
    transformed = transform_task(extract)
    load_task(transformed)
    notify_task(transformed)
# Airflow выводит граф: extract → transform → [load, notify]

Конкретный запуск DAG за определённый интервал (data interval). DagRun имеет state (queued/running/success/failed), logical_date (в 2.x — execution_date), run_id. Создаётся scheduler-ом из timetable или dataset event.

-- В metadata DB:
SELECT dag_id, run_id, state, start_date, execution_date
FROM dag_run
WHERE dag_id = 'orders_etl'
ORDER BY start_date DESC LIMIT 5;

Конкретный запуск конкретного task в конкретном DagRun. Имеет полный жизненный цикл состояний: none → scheduled → queued → running → success/failed/retry/deferred. Хранится в таблице task_instance metadata DB.

# TaskInstance состояния:
# none — deps не satisfied
# scheduled — все deps OK, выбран scheduler-ом
# queued — отдан executor-у
# running — выполняется на worker
# success / failed / up_for_retry / deferred

Компонент Airflow, отвечающий за выполнение tasks. В 2.x: LocalExecutor, CeleryExecutor, KubernetesExecutor, CeleryKubernetesExecutor, LocalKubernetesExecutor. С 2.10 — Multiple Executors одновременно (AIP-61).

# airflow.cfg
[core]
executor = LocalExecutor,CeleryExecutor  # multiple executors (AIP-61, 2.10+)

# Task с явным executor:
@task(executor='CeleryExecutor')
def heavy_task():
    pass

Главный процесс Airflow. Парсит DAGs (через DagFileProcessor pool внутри scheduler-process в 2.x), создаёт DagRun-ы по timetable, переводит TaskInstance scheduled → queued, кидает их в executor. Работает в HA режиме с 2.0+ через PostgreSQL row-level locks.

# Запуск scheduler standalone
airflow scheduler

# В Docker:
docker compose exec airflow-scheduler airflow scheduler

# Проверить heartbeat в metadata DB:
SELECT id, job_type, state, latest_heartbeat
FROM job WHERE job_type='SchedulerJob';

В 2.x — пул дочерних процессов (multiprocessing.Process) ВНУТРИ scheduler-процесса. Парсит .py файлы DAGs, сериализует их в БД (таблица serialized_dag). Можно опционально вынести в standalone process через `airflow dag-processor`. В 3.x стал mandatory отдельным процессом (AIP-66).

# airflow.cfg
[scheduler]
parsing_processes = 2  # parallel DagFileProcessor (default)
min_file_process_interval = 30

# Standalone:
airflow dag-processor

Отдельный процесс с asyncio event loop, держащий тысячи активных triggers для deferrable operators (AIP-40 stable с 2.5). Когда task вызывает self.defer(), trigger сериализуется и передаётся triggerer-у; событие просыпает task без блокировки worker-slot.

# Запуск:
airflow triggerer

# Custom trigger:
class MyTrigger(BaseTrigger):
    async def run(self):
        while not ready:
            await asyncio.sleep(1)
        yield TriggerEvent(data)

В Airflow 2.x — Flask application + Flask-AppBuilder (FAB) + Gunicorn. Server-rendered UI на Jinja templates. REST API v1 через flask-restful. Auth pluggable через FAB. В Airflow 3.x заменён на FastAPI + React UI.

# airflow.cfg
[webserver]
web_server_port = 8080
workers = 4                       # Gunicorn workers
worker_class = sync
worker_refresh_interval = 6000    # restart on big DagBag

В Scheduler — участок main loop (с Airflow 2.0+), где TI переводятся из scheduled в queued state с одновременной проверкой concurrency и pool limits. Защищён row-level lock на pool slots (SELECT * FROM slot_pool FOR UPDATE NOWAIT) — гарантирует, что одновременно только один scheduler в section, обеспечивая HA через RDBMS lock без Raft/ZK.

-- В Postgres логах:
LOG: process X waiting for ShareLock on transaction Y
  on relation slot_pool of database airflow

-- Это значит scheduler-1 в critical section, scheduler-2 ждёт

TaskInstance в состоянии running, чей heartbeat от worker не обновлялся дольше scheduler_zombie_task_threshold (default 300s). Причины: worker OOM-killed, node spot termination, network partition. Scheduler помечает их failed/up_for_retry.

# Конфиг:
[scheduler]
scheduler_zombie_task_threshold = 300

# Мониторинг:
SELECT dag_id, task_id, latest_heartbeat,
       now() - latest_heartbeat AS gap
FROM task_instance
WHERE state='running'
  AND latest_heartbeat < now() - interval '300 seconds';

TaskInstance в состоянии queued/running, чей scheduler/worker умер. Живой scheduler через adopt_or_reset_orphaned_tasks пытается забрать их (try_adopt_task_instances у executor-а) или сбросить в state=None для рестарта.

# Внутри SchedulerJob:
UPDATE job SET state='failed'
WHERE job_type='SchedulerJob'
  AND state='running'
  AND latest_heartbeat < now() - 30s;
# Затем — adopt их TI или reset

В Airflow 2.4+ (AIP-48) declarative data object с identifier. DAGs могут produce datasets (через outlets) и consume их (через schedule=[dataset]). Когда dataset обновлён, downstream DAGs автоматически triggers. В Airflow 3.x переименовано в Assets с расширениями.

from airflow import Dataset
from airflow.decorators import dag, task

my_dataset = Dataset('s3://bucket/data.parquet')

@dag(schedule=[my_dataset])
def consumer(): ...

@dag(schedule='@daily')
def producer():
    @task(outlets=[my_dataset])
    def write(): ...

Dynamic dataset references — для случаев, когда нельзя статически определить dataset URI. На runtime эмитим конкретный dataset event через context.

from airflow.datasets import DatasetAlias

my_alias = DatasetAlias('my-alias')

@task(outlets=[my_alias])
def dynamic_writer(context):
    context['outlet_events'][my_alias].add(
        Dataset(f's3://bucket/{path}')
    )

В Airflow 2.10+ возможность одновременно использовать несколько executors. Конфиг: executor = LocalExecutor,CeleryExecutor. Первый — default. Task может явно указать executor через @task(executor='CeleryExecutor'). В 3.2 добавлено team-scoping (AIP-67).

[core]
executor = LocalExecutor,CeleryExecutor,KubernetesExecutor

@task(executor='KubernetesExecutor')
def gpu_task():
    # запустится в k8s pod
    pass

Параметр Celery worker — сколько messages забирать из broker заранее. Значение по умолчанию 4 — long-task может занять 4 slots, блокируя другие. Best practice для mixed workload: worker_prefetch_multiplier = 1.

# airflow.cfg
[celery]
worker_prefetch_multiplier = 1  # против starvation
worker_concurrency = 16
worker_autoscale = 16,8  # min/max
broker_url = redis://redis:6379/0
result_backend = db+postgresql://airflow:...@postgres/airflow

В Airflow 2.x hybrid executor: по queue task выбирается путь. queue='kubernetes' → KubernetesExecutor (изоляция, slow start), всё остальное → CeleryExecutor (быстрый, shared workers). Также LocalKubernetesExecutor (Local + K8s).

[core]
executor = airflow.executors.celery_kubernetes_executor.CeleryKubernetesExecutor

# Task на K8s:
@task(queue='kubernetes')
def gpu_task(): ...

# Task на Celery:
@task
def light_task(): ...

Симметричный encryption Airflow для Connections и Variables. AES-128-CBC + HMAC-SHA256. Ключ в airflow.cfg fernet_key. Key rotation через несколько ключей: новые шифруются первым, старые расшифровываются любым.

# Генерация ключа:
python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())'

# Key rotation:
export AIRFLOW__CORE__FERNET_KEY='new_key,old_key'

# Re-encrypt:
airflow rotate-fernet-key

В Airflow 2.8+ новый 'правильный' способ слушать events lifecycle (task started/success/failed, dataset created/changed). Раньше использовали on_failure_callback per task. Listener видит ВСЕ events системы — идеально для generic alerting, audit logging, OpenLineage emission.

from airflow.listeners import hookimpl

class MyListener:
    @hookimpl
    def on_task_instance_running(self, previous_state, task_instance, session):
        send_to_slack(f'Task {task_instance.task_id} started')

Open standard для emission data lineage events. Airflow provider apache-airflow-providers-openlineage (стабилизирован с 2.6) автоматически эмитит OL events для каждого task (job start/complete/fail) с метаданными (SQL, source/destination, schema). Бэкенды: Marquez, DataHub, Atlan.

# airflow.cfg
[openlineage]
transport = {"type":"http","url":"http://marquez:5000"}
namespace = production

# Все стандартные operators emit-ят OL events автоматически

Operators, которые могут передать управление triggerer (async event loop) вместо блокировки worker slot. С 2.2 experimental, с 2.5 stable. Worker вызывает self.defer(trigger=MyTrigger(), method_name='execute_complete') → state=deferred → освобождает slot.

from airflow.providers.standard.sensors.date_time import DateTimeSensorAsync

@dag(...)
def my_dag():
    DateTimeSensorAsync(
        task_id='wait',
        target_time='2026-05-13T10:00:00'
    )  # ← НЕ блокирует worker slot

В Airflow 2.10+ полный stack OTel: metrics, traces, logs. Заменяет StatsD. Готовая интеграция с Datadog, Dynatrace, NewRelic, Grafana Cloud, Tempo. Ключевые метрики: scheduler_loop_duration, executor.open_slots, triggerer.running_triggers.

# airflow.cfg
[metrics]
otel_on = True
otel_host = otel-collector
otel_port = 4317
otel_prefix = airflow
otel_interval_milliseconds = 60000