Компоненты: веб-сервер (интерфейс и API), планировщик (scheduler), который разбирает файлы DAG, создаёт запуски по расписанию и ставит задачи в очередь, исполнители (executor: Local, Celery, Kubernetes) и воркеры, которые выполняют задачи, база метаданных (состояния запусков, XCom, соединения), триггерер для отложенных операторов. DAG это Python-файл с объектом DAG (id, расписание, start_date, параметры по умолчанию) и задачами (операторами), связанными зависимостями >>; граф должен быть ацикличным.
Планировщик при разборе файла проверит граф и отклонит DAG с ошибкой Cycle detected: DAG не появится в списке или будет помечен как сломанный, ни одна задача не запустится. Направленный ацикличный граф это требование, а не рекомендация, иначе невозможно определить порядок выполнения.
Расписание задаётся schedule (cron-строка, timedelta, пресеты вроде @daily, или datasets). Запуск с логической датой start_date происходит в конце интервала: при start_date = 16 июля и расписании ежедневно в 10:00 первый запуск произойдёт 17 июля в 10:00 с логической датой 16 июля, потому что интервал данных за 16-е считается завершённым только тогда. Параметр catchup решает, догонять ли пропущенные интервалы. Минимальный интервал формально любой (хоть минута), но на практике ограничен временем разбора DAG и ресурсами планировщика; для частых запусков берут сенсоры с отложенным режимом или стриминг.
dag_id, schedule, start_date, end_date, catchup, default_args (владелец, ретраи, retry_delay, уведомления, depends_on_past), max_active_runs, max_active_tasks, tags, params для ручного запуска, dagrun_timeout, on_failure_callback. В задачах важны retries, pool, priority_weight, trigger_rule, execution_timeout.
Scheduler решает, что и когда запускать, Executor решает, как и где задача будет выполнена: ставит её в очередь, следит за слотами параллелизма и пулами, передаёт воркеру и принимает статус. SequentialExecutor выполняет по одной задаче, работает с SQLite и годится только для разработки. LocalExecutor запускает задачи процессами на той же машине, что и планировщик, нужна PostgreSQL или MySQL. CeleryExecutor раздаёт задачи воркерам через брокер (Redis, RabbitMQ) и масштабируется добавлением воркеров. KubernetesExecutor поднимает под на каждую задачу с нужным образом и ресурсами, ничего не простаивает, но старт задачи дольше. Масштабируют переходом на Celery или Kubernetes, настройками parallelism, max_active_tasks_per_dag, пулами для внешних систем и несколькими планировщиками.
Catchup это параметр DAG: при catchup=True планировщик после включения DAG автоматически создаёт запуски для всех пропущенных интервалов от start_date до текущего момента, строго по порядку. Backfill это ручная команда airflow dags backfill -s ... -e ...: запуски за выбранный период, можно перезапускать уже существующие, ограничиться задачами по regex, игнорировать зависимости. На практике для продовых DAG ставят catchup=False, чтобы случайно не запустить сотни прогонов, а историю догружают backfill-ом осознанно, и пишут задачи идемпотентными по интервалу, чтобы перезапуск за период был безопасен.
Когда готовых задач больше, чем свободных слотов, Executor берёт задачи с большим весом первыми. priority_weight задаёт базовый вес задачи, а weight_rule способ его подсчёта: downstream (по умолчанию) суммирует веса всех задач ниже по графу, поэтому начало длинной цепочки важнее, upstream суммирует веса выше по графу, absolute берёт собственный вес без суммирования, что дешевле для больших DAG. Используют, чтобы критичный пайплайн не ждал в очереди за сотней мелких задач при общем пуле.
Оператор это шаблон задачи: экземпляр оператора в DAG становится задачей. Базовые: PythonOperator и @task для кода, BashOperator, SQLExecuteQueryOperator/PostgresOperator для SQL, SparkSubmitOperator, KubernetesPodOperator для изолированного окружения, EmptyOperator для структуры, BranchPythonOperator для ветвления, TriggerDagRunOperator. Параметры: task_id, retries, pool, trigger_rule, execution_timeout, шаблонизируемые поля с Jinja ({{ ds }}). Честный ответ называет два-три оператора и объясняет, почему SQL в базе выполняется оператором, а не Python-кодом с курсором.
Сенсор это оператор, который ждёт условия: появления файла в S3, партиции в Hive, записи в таблице, завершения задачи в другом DAG (ExternalTaskSensor), времени. Режимы poke (занимает слот воркера) и reschedule (освобождает между проверками), отложенные сенсоры через триггерер. Хук это обёртка над соединением к внешней системе (PostgresHook, S3Hook, HttpHook): берёт реквизиты из Connections и даёт методы; операторы внутри используют хуки, и в своём @task их тоже удобно брать вместо ручных подключений.
XCom передаёт маленькие значения между задачами через базу метаданных: имя файла, число строк, дату; @task возвращает значение, и оно становится XCom автоматически. Для данных XCom не подходит: всё хранится в базе Airflow, большие объёмы её кладут; данные передаются через S3/таблицу, а через XCom только ссылка. Variables это глобальные настройки (пути, лимиты), секреты в Connections или внешнем секрет-бэкенде. Динамические DAG и генерация задач из конфига (expand для dynamic task mapping) это способ не писать сто одинаковых задач руками.
Логическая дата запуска (logical_date, ds) это начало интервала данных, а не момент запуска: задача за 1 апреля запускается 2 апреля, и в код нужно передавать data_interval_start/data_interval_end, а не datetime.now(). Часовые пояса: start_date без tz считается UTC. catchup=True при включении старого DAG запускает все пропущенные интервалы. Изменение start_date или schedule у существующего DAG путает историю запусков. Лечится правилом: вся логика периода берётся из контекста запуска параметром.
Сенсор ждёт события: файла, партиции, ответа API. В режиме poke он занимает слот воркера всё время ожидания и проверяет условие каждые poke_interval секунд: подходит для коротких ожиданий. В режиме reschedule между проверками задача освобождает слот и возвращается в очередь, воркер не занят, но у планировщика больше работы. Deferrable-операторы и сенсоры (.defer()) отдают ожидание отдельному процессу triggerer, который держит тысячи асинхронных ожиданий в одном event loop; воркер не задействован вообще, это правильный выбор для долгих ожиданий. Общие параметры: timeout (по умолчанию семь дней), exponential_backoff, soft_fail.
При истечении timeout сенсор по умолчанию падает в failed, DAG помечается неудачным и приходят алерты. С soft_fail=True задача становится skipped, и зависимые задачи по умолчанию тоже пропускаются, а DAG завершается успешно. Это для необязательных данных: бонусный файл, выгрузка, которой не бывает в выходные, один из нескольких альтернативных источников. Для критичных входов оставляют soft_fail=False, иначе пропуск скроет реальную проблему. Чтобы после пропущенного сенсора всё же выполнить задачу, ей ставят trigger_rule="all_done" или none_failed, а ветвление делают через BranchPythonOperator.
Любое необработанное исключение в python_callable переводит задачу в failed, но принято бросать AirflowException с понятным сообщением: оно попадает в лог и в алерт. При retries > 0 задача сначала уйдёт в up_for_retry и будет перезапущена. AirflowFailException игнорирует retries и сразу помечает задачу проваленной: для случаев, когда повтор бессмыслен, например неверная конфигурация или невалидные входные данные. AirflowSkipException переводит задачу в skipped, когда делать нечего, и это не ошибка.
Dynamic task mapping в Airflow 2.3+: оператор.partial(общие_параметры).expand(параметр=список) создаёт на этапе выполнения по экземпляру задачи на каждый элемент списка, причём список может быть результатом предыдущей задачи через XCom. Так обрабатывают переменный набор файлов, таблиц или дат параллельно без цикла в коде DAG и без пересборки графа. Отличие от генерации DAG кодом: там граф фиксируется при разборе файла, а здесь число задач известно только в рантайме; в UI mapped-задачи видны как одна задача с индексами.
XCom хранится в базе метаданных, поэтому предел зависит от неё: в MySQL это 64 КБ на значение, в PostgreSQL формально до 1 ГБ, в SQLite 2 ГБ. Но XCom задуман для метаданных: путь к файлу, число строк, идентификатор запуска. Таскать через него датафреймы значит нагружать базу планировщика и сериализацию. Для больших объёмов данные кладут в S3 или таблицу и передают только ссылку, либо подключают custom XCom backend, который сам пишет значения в объектное хранилище.
Задачи падают и перезапускаются (retry, backfill, ручной перезапуск), и повторный запуск за тот же интервал должен дать тот же результат без дублей. Обеспечивается: период берётся из контекста запуска, а не из текущего времени; запись заменяет период (DELETE+INSERT в транзакции, замена партиции, upsert по ключу); временные файлы именуются по интервалу; внешние вызовы с ключом идемпотентности. Признак неидемпотентной задачи: INSERT без очистки и now() в коде.
Типичные проблемы и решения: медленный разбор DAG-файлов (тяжёлый код на верхнем уровне файла, обращения к базе при импорте; выносить в задачи, кэшировать конфиг), очередь задач (пулы и приоритеты, больше воркеров, Celery/Kubernetes executor), сенсоры, занимающие слоты (режим reschedule и отложенные операторы), раздутая база метаданных (чистка XCom и логов), один гигантский DAG (разбить, dynamic task mapping), падения из-за зависимостей между DAG (datasets вместо сенсоров).
Пример: DAG загрузки заказов из API в хранилище, ежедневно в 03:00. Задачи: сенсор или проверка доступности источника → выгрузка инкремента за интервал в S3 (raw, JSON Lines) → загрузка в staging Postgres/Greenplum → SQL-преобразование в core с идемпотентной заменой периода → проверки качества → обновление витрины → уведомление. Ретраи на сетевых шагах, пул на подключения к базе, алерт на падение, параметр даты для перезапуска за прошлый день. Роль разработчика: пишет DAG и код, деплой через git и CI; админ Airflow настраивает Connections, пулы, роли, что в маленьких командах делает тот же инженер.
retries и retry_delay задаются в default_args на весь DAG или у конкретной задачи, где переопределяют общие; retry_exponential_backoff и max_retry_delay разводят повторы по времени. Перезапускается только упавшая задача, не весь DAG. На события вешают колбэки: on_failure_callback, on_retry_callback, on_success_callback, sla_miss_callback для уведомлений в Slack или Telegram и для компенсирующих действий. trigger_rule у зависимых задач задаёт реакцию на статусы предков: all_success по умолчанию, all_done, one_failed, none_failed.
Первый уровень: файл DAG должен импортироваться без ошибок и быстро, для этого есть тест на DagBag с проверкой import_errors и ограничения на время разбора. Юнит-тесты проверяют функции из python_callable отдельно от Airflow. airflow tasks test dag task date выполняет одну задачу без записи в базу, airflow dags test прогоняет DAG целиком за дату. Структуру графа проверяют на зависимости и циклы. Версионирование: DAG-файлы лежат в Git, выкатываются через CI/CD в папку dags (или через git-sync в Kubernetes), а сам Airflow историю версий DAG не хранит до версии 3, поэтому изменения логики за прошлые даты нужно продумывать с backfill.
Connection описывает подключение к внешней системе: тип, хост, порт, логин, пароль, extra в JSON; его используют хуки и операторы по conn_id, пароль хранится зашифрованным ключом Fernet. Variable это именованное значение для параметров DAG: список таблиц, дата, флаг; читается через Variable.get, в шаблонах {{ var.value.x }}. Чтение Variable внутри кода DAG на верхнем уровне делает запрос к базе при каждом разборе файла, поэтому их читают внутри задач или через шаблоны. Для секретов подключают secrets backend (HashiCorp Vault, AWS Secrets Manager, Kubernetes Secrets): Airflow ищет connection и variable сначала там, потом в базе.