Apache Airflow: как создать первый DAG на Python

Допустим, у вас два шага обработки: сначала получить данные, затем посчитать результат. В обычном скрипте достаточно вызвать две функции подряд. Но когда процесс нужно запускать повторно, разбираться с ошибками и видеть, на каком шаге всё остановилось, хочется управлять этими действиями по отдельности. Для этого и пригодится Apache Airflow.

Чтобы понять его устройство, не будем сразу поднимать сложный конвейер с базой данных и очередями. Напишем DAG с двумя задачами: load_numbers подготовит список [1, 2, 3], а calculate_total получит его и вычислит сумму. Затем запустим процесс вручную и найдём total=6 в журнале второй задачи.

Всё, что потребуется, — знакомство с Python и терминалом. Пример рассчитан на Apache Airflow 3.3.2 и Python 3.12 в Linux, macOS или WSL2. В конце вы будете понимать, что именно делает DAG-файл, зачем нужен TaskFlow и как убедиться, что Airflow выполнил обе задачи.

Что такое Apache Airflow и DAG

Apache Airflow помогает управлять процессами из нескольких шагов. Вы задаёте задачи, порядок их выполнения и условия запуска; Airflow отслеживает отдельные выполнения и позволяет найти ошибку в конкретном шаге. Это удобно для обработки данных, регулярных отчётов и других работ, где важна последовательность действий.

Основа такого процесса — DAG, сокращение от Directed Acyclic Graph, то есть направленного ациклического графа. DAG описывает состав задач и зависимости между ними; содержимое задач задаёт Python-код. В нашем примере связь одна: сначала load_numbers, затем calculate_total. Направление показывает, кто от кого зависит; циклов быть не должно, иначе нельзя определить порядок выполнения. Подробнее устройство графа разобрано в документации Airflow.

Здесь полезно сразу различать три понятия:

• DAG — описание всего процесса: его задачи и связи.
• Task — отдельная задача внутри DAG. Мы создадим две такие задачи из обычных Python-функций.
• DagRun — одно конкретное выполнение DAG. Внутри него у каждой задачи есть свой экземпляр — Task Instance, со статусом и журналом.

Представьте, что вы запускаете наш процесс дважды. Описание DAG остаётся одним и тем же, зато появляются два DagRun, и каждый хранит собственные результаты задач. Поэтому Airflow позволяет отличить «процесс описан правильно» от «в этот раз процесс выполнился успешно».

Установите Airflow для первого запуска

Airflow — это больше, чем библиотека с декораторами для Python. Чтобы запустить задачи и посмотреть их состояния, ему нужны служебная база данных и компоненты выполнения. Для обучения всё это можно поднять на одной машине без отдельного PostgreSQL или очереди сообщений.

Создайте рабочий каталог и виртуальное окружение Python 3.12. Команды ниже подходят для Linux и macOS; на Windows используйте WSL2:

mkdir airflow-first-dag
cd airflow-first-dag
python3.12 -m venv .venv
source .venv/bin/activate
export AIRFLOW_HOME="$PWD/.airflow"
CONSTRAINT_URL=https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt
python -m pip install "apache-airflow==3.3.2" --constraint "$CONSTRAINT_URL"
airflow version

Последняя команда должна показать 3.3.2. Файл constraints закрепляет совместимые версии библиотек для выбранного выпуска Airflow и Python. Поэтому мы используем его вместе с конкретной версией пакета, как рекомендует официальная инструкция по установке. Для установки понадобится интернет.

Обратите внимание на AIRFLOW_HOME: здесь Airflow будет хранить настройки, служебную базу и каталог с DAG-файлами. Мы задали путь внутри учебного проекта, чтобы он не смешивался с другими установками. Во втором терминале эту переменную нужно будет задать снова.

airflow standalone запускает локальную конфигурацию Airflow для разработки и обучения. Эту команду выполним после создания DAG. Согласно официальному Quick Start, интерфейс будет доступен по адресу http://localhost:8080. Airflow создаст локальную учётную запись; если пароль не показан в терминале, найдите его в $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated. Это секрет вашего локального окружения — не добавляйте его в код или публикации.

Режим standalone рассчитан на знакомство с Airflow. Для рабочего сервиса потребуются отдельные решения по развёртыванию, хранению данных и безопасности.

Напишите первый DAG на Python

**В Airflow 3 публичные декораторы dag и task импортируются из airflow.sdk. С помощью @dag мы описываем весь процесс, а с помощью @task — каждый его шаг. Такой подход называется TaskFlow API. Он позволяет писать логику задач как обычные функции Python; пример есть в официальном руководстве TaskFlow.

Создайте в рабочем каталоге файл first_dag.py и добавьте код:

"""Minimal manually triggered TaskFlow DAG for Apache Airflow 3.3.2."""

from __future__ import annotations

import pendulum
from airflow.sdk import dag, task


@dag(
    dag_id="first_python_dag",
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["tutorial"],
)
def first_python_dag() -> None:
    @task
    def load_numbers() -> list[int]:
        return [1, 2, 3]

    @task
    def calculate_total(numbers: list[int]) -> int:
        total = sum(numbers)
        print(f"total={total}")
        return total

    calculate_total(load_numbers())


first_python_dag()

Разберём важные строки. Декоратор @dag задаёт имя first_python_dag и настройки запуска. schedule=None оставляет DAG без автоматического расписания: первый запуск сделаем вручную. Дата start_date записана с часовым поясом UTC, чтобы не зависеть от настроек компьютера. catchup=False станет важен, если позже добавить расписание: он запрещает автоматически создавать накопившиеся пропущенные запуски.

Две функции пока выглядят как обычный Python: первая возвращает числа, вторая печатает сумму. Отличие проявится при запуске — Airflow выполнит их как отдельные задачи, а не как один непрерывный скрипт.

Наконец, строка first_python_dag() в конце файла создаёт объект DAG, когда Airflow читает модуль. В этот момент сумма ещё не вычисляется: тела задач выполняются позднее, при запуске процесса.

Переместите файл туда, где Airflow ищет определения:

mkdir -p "$AIRFLOW_HOME/dags"
cp first_dag.py "$AIRFLOW_HOME/dags/first_dag.py"
airflow config get-value core dags_folder

Проверьте, что последняя команда указывает на $AIRFLOW_HOME/dags. Если путь другой, используйте каталог из фактической конфигурации Airflow.

Для тех, кто захочет изучить проверки глубже, есть полный пример с тестами. Но для первого запуска достаточно кода выше.

Как связаны две задачи TaskFlow

Вызов calculate_total(load_numbers()) задаёт связь двух задач через результат первой. При построении графа Airflow создаёт два объекта задачи: load_numbers находится выше по зависимости (upstream), calculate_total — ниже (downstream).

На уровне обычного Python запись выглядит как вызов одной функции внутри другой. Однако декоратор @task изменяет семантику вызова при объявлении DAG: строится граф, а реальное вычисление происходит позже. Отдельное выражение load_numbers() >> calculate_total() для передачи аргумента здесь не нужно.

Когда первая задача возвращает список [1, 2, 3], TaskFlow может передать его следующей через XCom — механизм обмена небольшими сериализуемыми результатами между экземплярами задач. Документация XCom предупреждает, что большие таблицы и файлы следует хранить иначе. Для учебного списка из трёх чисел XCom подходит.

Если убрать выражение calculate_total(load_numbers()) из объявления DAG, обе декорированные функции останутся в файле, но не сформируют этот граф задач. Поэтому в проверке важны не только имена функций, но и созданные связи.

Запустите DAG и проверьте результат

После копирования first_dag.py оставьте первый терминал в виртуальном окружении и запустите:

airflow standalone

Дайте локальным компонентам завершить начальную настройку. Откройте в браузере http://localhost:8080 и войдите с локальными учётными данными. Не размещайте пароль в публичных логах и скриншотах.

Откройте второй терминал в том же рабочем каталоге, повторно активируйте окружение и установите тот же AIRFLOW_HOME:

source .venv/bin/activate
export AIRFLOW_HOME="$PWD/.airflow"
airflow dags list
airflow dags list-import-errors
airflow tasks list first_python_dag

В списке DAG должен появиться first_python_dag, ошибок импорта для него быть не должно. Вывод списка задач должен содержать load_numbers и calculate_total. Это ещё проверка обнаружения и структуры, не выполненный расчёт.

Успешная загрузка DAG подтверждает разбор определения, но ещё не подтверждает выполнение задач. Для одного тестового запуска используйте команду Airflow 3.3.2 с явным путём к файлу:

airflow dags test first_python_dag --dagfile-path "$AIRFLOW_HOME/dags/first_dag.py"

Параметр --dagfile-path и смысл airflow dags test зафиксированы в официальном справочнике CLI. Команда предназначена для одного тестового DagRun. В ожидаемом успешном случае обе задачи перейдут в состояние success, а в журнале calculate_total будет строка:

total=6

После успешного выполнения обе Task Instance имеют статус success, а журнал calculate_total содержит total=6. Это подтверждает не только загрузку определения, но и передачу списка между задачами во время DagRun.

После выполнения найдите DagRun и журнал в интерфейсе Airflow. Если используете отдельный ручной запуск через интерфейс, проверьте его собственные состояния и логи: результат тестовой команды и состояние другого запуска не следует смешивать. Названия, коды завершения и лог на своей машине фиксируйте из реального запуска, а не из ожидаемого текста руководства.

Сама арифметика проверяется независимо: сумма 1 + 2 + 3 равна 6. Но только работающий Airflow подтвердит, что задачи действительно выполнились в нужном порядке и результат был передан через XCom.

Если DAG не появился или завершился с ошибкой

В Airflow полезно искать проблему в том порядке, в котором проходит запуск: сначала программа должна найти DAG, затем разобрать его задачи и только после этого выполнить их.

DAG нет в списке. Проверьте AIRFLOW_HOME в обоих терминалах и выполните airflow config get-value core dags_folder. В этом каталоге должен лежать first_dag.py. Если файл на месте, посмотрите вывод airflow dags list-import-errors — он помогает отличить ошибку импорта от ошибки расположения файла.

Airflow не может импортировать модуль. При ModuleNotFoundError сначала проверьте, активировано ли виртуальное окружение: python --version и airflow version должны показывать выбранный Python и Airflow 3.3.2. Исправив окружение, снова выполните airflow dags list.

В DAG нет второй задачи. Вспомните разницу между определением функции и её вызовом. Одна только функция с @task ещё не образует шаг графа. Проверьте, что внутри first_python_dag осталась строка calculate_total(load_numbers()). Затем посмотрите список задач или граф в интерфейсе.

DAG загрузился, но результат не появился. Откройте журнал именно упавшей Task Instance. Если ошиблась функция, сообщение будет в её журнале; если проблема в служебной базе или запуске, проверьте терминал с airflow standalone. Не спешите удалять metadata DB: так вы потеряете сведения, которые помогают понять причину. В руководстве по отладке описаны требования к запуску и сериализации DAG.

После исправления повторите действия по порядку: обнаружение DAG, две задачи, тестовый запуск, их состояния и строка total=6. Одна лишь проверка синтаксиса командой python -m py_compile не говорит, что задачи действительно отработали.

Что вы получили

Теперь у вас есть первый DAG на Python: load_numbers возвращает три числа, а calculate_total получает их и печатает total=6. Вы знаете, где Airflow ищет файл, как появляются задачи из @task и почему их порядок определяется связью в графе.

Самая полезная привычка после этого упражнения — смотреть на конкретный запуск, а не только на исходный Python-файл. DAG может успешно загрузиться, но задача всё равно может упасть во время выполнения. Состояния задач и их журналы показывают, на каком шаге возникла проблема.

Для следующего опыта замените [1, 2, 3] чтением небольшого локального файла и сохраните результат в отдельный файл. При этом продумайте, что произойдёт при повторном запуске: не создаст ли он дубликаты и сможет ли безопасно начаться заново. Именно такие вопросы становятся важны, когда учебный DAG превращается в реальный процесс обработки данных. Для производственного окружения дополнительно понадобятся настройки безопасности, хранения состояния и исполнения задач.

Источники

• Официальные анонсы релизов и подтверждение версии 3.3.2
• DAG, Task, зависимости и режим запуска
• TaskFlow API, @dag/@task, передача результатов
• Передача небольших результатов между задачами через XCom
• Первый DAG, структура, импорт и тестирование
• Установка и локальный airflow standalone
• Установка через pip/uv с constraints
• Официальный constraints-файл Airflow 3.3.2 для Python 3.12; вход установки, а не свидетельство успешного запуска
• airflow dags test, tasks list и диагностика через CLI
• Условия запуска dag.test и требования к базе/обнаружению DAG
• Публичный Task SDK для Airflow 3
• Полный воспроизводимый пример, pytest и настоящий DagRun в закреплённом публичном commit