Airflow DAG: как собрать зависимости и проверить запуск

Airflow DAG — описание задач, зависимостей и правил запуска рабочего потока в Apache Airflow. Начните с маленького графа без нейросети: источник тестовых данных передаёт список задаче подсчёта, а вы проверяете порядок и состояние каждого выполнения. Такой каркас полезен, когда вход, ожидаемый результат и владелец последующей обработки уже определены.

Каркас графа

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

Каркас для документации Airflow 3 с интерфейсом airflow.sdk выглядит так:

from airflow.sdk import dag, task

@dag(schedule=None)
def row_check():
    @task
    def source():
        return [1, 2, 3]

    @task
    def count(rows):
        if len(rows) != 3:
            raise ValueError("unexpected row count")
        return len(rows)

    count(source())

row_check()

Файл с графом разместите в настроенном каталоге Dag вашей среды и проверьте, что Airflow его обнаружил. Для установленной версии сверьте путь импорта и способ загрузки: в актуальной документации Airflow 3 исходные файлы читаются из Dag bundles. Появление графа в интерфейсе — отдельный приёмочный шаг: ошибка импорта способна остановить загрузку.

Соседний материал об Airflow и операторах для ИИ начинает с отдельного шага модели. Этот разбор о предшествующем основании: простой граф должен запускаться, показывать состояния и давать проверяемый результат до добавления ИИ. Общая логика сценариев оркестрации также раскрыта в статье о сценариях n8n, но команды и модель выполнения здесь принадлежат Airflow.

Зависимость и вход

Связь между source и count отвечает на конкретный вопрос: откуда задача подсчёта получает список. В TaskFlow передача результата функции одновременно задаёт зависимость. При более сложной схеме зависимости можно объявлять явно; документация Airflow показывает операторы >> и <<. Направление стрелки важно: задача после стрелки ждёт верхнюю задачу, а порядок вызовов Python отражает лишь часть отношений в графе.

1. Опишите вход каждой задачи и ожидаемый выход. Для учебного графа вход source задан в коде, а count получает список.
2. Назовите владельца данных и формат результата до замены учебного списка реальным источником.
3. Проверьте в Graph View направление связи и отсутствие случайной ветки, которая минует проверку.
4. Запустите граф вручную в тестовой среде и сопоставьте возвращённое значение с исходным списком.

Небольшой список уместен для проверки каркаса. Рабочие объёмы данных требуют другого решения по передаче и хранению; пример с TaskFlow показывает механику зависимости; крупные таблицы между задачами передавайте через подходящее хранилище. Для реального процесса выберите устойчивый идентификатор набора данных, проверяемый источник и отдельный результат выполнения. Тогда повторный запуск можно связать с тем же входом и разобраться, откуда появился другой ответ.

Права описывайте рядом с входами. Автор Dag может менять код и зависимости, оператор запускает согласованный граф, а владелец данных разрешает доступ к источнику. Если следующий этап меняет рабочую таблицу, серверный слой проверяет полномочия на запись, а ответственный человек утверждает спорный результат. Наличие связи в Airflow само по себе недостаточно для разрешения изменять внешнюю систему.

Расписание и запуск

Параметр schedule=None оставляет учебный граф для ручного запуска. По официальной документации Airflow граф также может запускаться по заданному расписанию или через API. Расписание определяет момент создания Dag Run; смысл входных данных и полномочия исполнителя задаются отдельно. Сначала выясните, когда исходный набор готов и кто подтверждает его полноту; только затем выбирайте правило повторения.

В Airflow каждый запуск графа получает собственный Dag Run и экземпляры задач. У запланированного запуска есть интервал данных и логическая дата, которые могут отличаться от фактического времени начала выполнения. Это важно при регулярной обработке: запрос «данные за период» нельзя строить по случайному времени старта воркера. Для учебного ручного запуска достаточно зафиксировать выбранный вход и посмотреть журнал конкретного Dag Run.

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

Состояния и сбои

После запуска откройте состояния отдельных Task Instances, а затем состояние Dag Run. Руководство по задачам Airflow различает ожидание, очередь, выполнение, успех, ошибку и пропуск. Успех задачи означает завершение её кода без ошибки; для оценки исходных бизнес-данных нужна отдельная проверка. Ошибка count в учебном графе означает, что условие о длине списка было нарушено, поэтому смотреть нужно вход и журнал именно этой задачи.

• Если Dag отсутствует в интерфейсе, проверьте ошибку импорта, каталог загрузки и наличие объекта Dag в файле.
• Если source завершился, а count ждёт, проверьте зависимость и состояние конкретного Dag Run.
• Если задача перезапущена, сопоставьте повтор с тем же идентификатором входного набора и проверьте побочные записи.
• Если итоговый статус кажется успешным при неполном результате, разберите условия завершения конечных задач и бизнес-проверку.

В документации Dag Run состояние графа зависит от конечных задач. Поэтому зелёный итог в интерфейсе нельзя использовать как единственное доказательство качества файла или отчёта. Приёмка требует отдельного сравнения результата с входом и критерием владельца процесса. Сервер проверяет полномочия перед записью во внешнюю систему; модель, если она появится позже, даёт черновик, а спорную оценку подтверждает человек.

На первом этапе достаточно явного журнала ошибок и ответственного за повтор. Для производственного маршрута согласуйте, кто изучает сбой, что разрешено перезапускать и как избежать повторного действия во внешнем сервисе. Эти вопросы возникают до добавления ИИ-шага. Если у команды уже есть сложный сценарий проверок агента, его границы подробно разобраны в статье о тестировании ИИ-агентов.

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

Приёмка и ИИ

Приёмку учебного Dag проведите на зафиксированном списке. Проверьте загрузку определения, связь source и count, ручной запуск, результат и понятную ошибку после намеренного изменения условия теста. Для каждой проверки сохраните идентификатор Dag Run, состояние задач и ссылку на журнал. Если условия изменили, верните учебный граф к исходному коду до следующей проверки. Такой маршрут показывает, что граф управляет порядком и состояниями.

После этого можно спроектировать ИИ-шаг между чтением входа и записью результата. Он получает подготовленный текст или ссылку на источник и возвращает предложение, например категорию обращения. Следующая задача проверяет строгую схему ответа и полномочия, а сотрудник подтверждает спорную классификацию перед использованием. Сам по себе оператор модели описан в соседней статье про Airflow operators; здесь важнее его место в цепочке и условия остановки при ошибке.

Для рабочей схемы зафиксируйте источник данных, период обработки, формат результата, права на запись, повторяемость задач и маршрут исключений. Техническое состояние Airflow сравнивайте с фактическим результатом в системе назначения. Если Dag Run успешен, а бизнес-артефакт отсутствует, возвращайтесь к задаче записи и проверяйте её контракт. Критерий годности заранее утверждает владелец процесса; спор оставляйте ответственному сотруднику.

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

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