Платформа на Airflow 3
Новый кластер Airflow 3.1 для хранилища данных: DAG приходят из Git без деплоя, каждый merge request проверяется в CI, модели dbt запускаются в отдельных контейнерах через Cosmos.
attributes
- role
- lead engineer
- source
- рабочий проект, код закрыт
Задача
Хранилище работало на Airflow 1.10: около 190 DAG и больше 500 подключений. DAG раскатывал один job в CI, который выполнял git pull по SSH на пяти ВМ. До merge ничего не проверялось, поэтому ошибка импорта обнаруживалась уже на серверах. Поддержка Airflow 1.10 закончилась в 2021 году.
Решение
Новый кластер на Airflow 3.1.5: два control-plane хоста со scheduler, DAG processor, API server, PostgreSQL и RabbitMQ и три Celery-воркера. Компоненты работают в контейнерах Podman как юниты systemd, кластер раскатывается через Ansible, образы собирает kaniko в GitLab CI.
- DAG bundles: два Git-репозитория подключены как GitDagBundle, Airflow отслеживает основную ветку и обновляется раз в 30 секунд. Версия git-провайдера зафиксирована: ранний релиз удалял .git из версии бандла, и каждая таска клонировала репозиторий заново
- CI для репозиториев DAG: ruff с отчетом GitLab Code Quality, проверка форматирования и проверка импорта, которая загружает DagBag и сериализует каждый DAG в боевом образе Airflow. Ошибки импорта и повторяющиеся dag_id блокируют merge. pre-commit запускает ruff при коммите и не пропускает коммиты напрямую в основную ветку
- dbt через astronomer-cosmos: команды dbt run, test и source выполняются в отдельных контейнерах на хостах воркеров, с аутентификацией Kerberos для Hive и Impala. Одна метрика - одна группа тасок: модели dbt, перенос из Hive в PostgreSQL и агрегаты. Перед расчетом gate-группы проверяют свежесть источников и наличие данных за дату
- Безопасные релизы: образы из merge request получают свои теги, а тег, по которому обновляется кластер, публикует только основная ветка
- Общий Python-пакет для кода DAG со своим CI: линтер, тесты, сборка wheel и версионированные релизы
Стек
Airflow 3.1.5 с CeleryExecutor, RabbitMQ, PostgreSQL 17 с WAL-G, Podman, Ansible, kaniko, GitLab CI, ruff, pre-commit, dbt-core 1.7 с адаптерами Hive, Impala и PostgreSQL, astronomer-cosmos 1.14.
Моя роль
Lead engineer: спроектировал кластер, доставку DAG и проверки в CI, настроил запуск dbt через Cosmos, веду перенос DAG с Airflow 1.10.
Результат
Airflow забирает изменения DAG из Git раз в 30 секунд после merge, без отдельного деплоя. Ошибки импорта и стиля находятся в merge request, а не на серверах. В проекте dbt около 200 моделей и 95 тестов. Перенос DAG с Airflow 1.10 продолжается.