Airflow DAG Platform

▶ Схема потока данных — самодостаточный HTML, открывается в браузере — без установки и сборки

Ситуация

Ручной отчёт по продажам — SQL-клиент, копирование агрегата в Google Sheets — давал задержки, ошибки копирования и никакой воспроизводимости. Пропущенный или упавший прогон оставался незамеченным, а новый отчёт приходилось собирать заново.

Задача

От меня требовалось заменить ручную выгрузку воспроизводимым и идемпотентным пайплайном: данные доходят до приёмника по расписанию без ручного вмешательства, падения видны сразу, а новый отчёт добавляется по общему шаблону, а не с нуля.

Действия

DAG-и

DAG	Расписание	Что делает	Приёмник
`daily_sales_report`	`@daily`	Агрегат продаж за сутки (`sale_date`, `product_id`, `revenue`, `orders`)	Google Sheets (или CSV-фолбэк)
`weekly_sales_summary`	`@weekly`	Сводка за неделю по `category` (`category`, `revenue`, `orders`)	CSV (Sheets опционально)

Каждый DAG строится по шаблону fetch → validate → write → record_metrics на @task-декораторах Airflow 3 SDK. Окно — полуинтервал [logical_date − N, logical_date); catchup=False, backfill — явной командой.

Переиспользуемые модули

dags/shared/
  config.py    # конфиг из Airflow Variables, без хардкода
  dq.py        # DQ-валидация перед записью
  alerts.py    # формат и callback алертов при падении
  metrics.py   # observability: запись в dag_metrics
  output.py    # CSV-райтер (dev-фолбэк), идемпотентная перезапись
  windows.py   # daily/weekly окна по logical_date

Стек и запуск

uv sync
uv run pytest                                 # тесты без Airflow runtime
uv run python scripts/generate_sales_data.py --days 7 --rows-per-day 10000
export AIRFLOW_HOME=$(pwd) && uv run airflow standalone   # UI :8080, admin/admin
uv run python scripts/metrics_report.py --days 7

Гейты CI

uv run ruff check .          # линт: E,W,F,I,B,C4,UP
uv run ruff format --check .
uv run mypy                  # dags/, scripts/, tests/
uv run pytest                # 64 теста, хуки замоканы

Результат

Платформа доводит два отчёта до приёмника по расписанию, каждый — отдельный версионируемый DAG. Качество держится на четырёх гейтах CI: ruff check, ruff format --check, mypy и pytest проходят на 64 тестах, которым не нужны ни Airflow-сервисы, ни живая БД (хуки замоканы). Конфигурация целиком вынесена в Airflow Variables с префиксом AIRFLOW_CFG_, а dev-режим output="file" работает без внешних API.

Ограничения

Источник данных синтетический (scripts/generate_sales_data.py), поэтому платформа проверена на демо-данных, а не на прод-потоке. Google Sheets как приёмник требует реального сервисного аккаунта; без него остаётся только CSV-фолбэк. Проект портфельный: надёжность поведения Airflow под нагрузкой на больших объёмах не измерялась.

Документация

GitHub → airflow

Частые вопросы

Почему отчёт — это DAG, а не скрипт?

DAG даёт версионирование, расписание, ретраи и метрики запусков из коробки; ручная выгрузка SQL-клиентом → копирование в Sheets не оставляет ни одного из этих следов.

Что проверяет DQ-валидация?

Заголовок результата, неотрицательные revenue и orders и пустой набор как отдельный NoDataError — до того, как данные уйдут в приёмник.

Как устроена observability?

Каждый успешный запуск пишет строку в dag_metrics (status, duration_s, rows_written, logical_date); провал фиксируется в on_failure_callback. scripts/metrics_report.py агрегирует success rate и p95.

Работает ли проект без внешних API?

Да: в dev-режиме AIRFLOW_CFG_sales_report_output=file пишет CSV в data/out/, а Google Sheets нужен реальный сервисный аккаунт.

Как добавить новый отчёт?

По общему шаблону переиспользования shared-модулей: новый DAG собирается из тех же config, dq, alerts, metrics и windows, без копирования логики.