Запустить DAG в Apache Airflow™

Инструкция описывает веб-интерфейс Apache Airflow™ 3.1.8. Для проверки можно использовать DAG hello_airflow из примера загрузки.

Перед началомПеред началом

Убедитесь, что кластер находится в состоянии Ready, DAG загружен и у вас есть учетные данные для входа.

Запустить вручнуюЗапустить вручную

1. Откройте веб-интерфейс Apache Airflow™ и перейдите в раздел Dags.
2. Найдите нужный DAG и откройте его. Если DAG приостановлен, включите его переключателем рядом с именем.
3. Нажмите Trigger. В открывшемся окне выберите Single Run.
4. Если DAG требует параметры запуска, заполните их. Для hello_airflow можно оставить значение message по умолчанию. Подтвердите запуск кнопкой Trigger.
5. Откройте вкладку Runs и выберите созданный запуск. Дождитесь его завершения.

Для hello_airflow ожидаемый результат — состояние success у запуска и задачи greet. При ошибке посмотрите логи задачи.

Повторный ручной запуск создает отдельный запуск DAG. Перед повтором DAG, который изменяет внешние данные, убедитесь, что повторная обработка допустима.

Элементы интерфейса описаны в документации Apache Airflow™.

Передать параметры запускаПередать параметры запуска

1. Для DAG hello_airflow нажмите Trigger → Single Run.
2. В поле message укажите Hello from parameters! и нажмите Trigger.
3. Откройте созданный запуск, задачу greet и вкладку Logs. В логе должно быть переданное сообщение; на вкладке XCom — возвращенная строка.

Поле появляется благодаря params в примере DAG. Параметр message имеет тип string; форма проверяет введенное значение. Параметры относятся к выбранному запуску и не изменяют значение по умолчанию в Python-файле.

Если DAG использует dag_run.conf, откройте Advanced Options и задайте Configuration JSON в соответствии с требованиями его автора. Не передавайте пароли и ключи доступа в параметрах запуска: для подключений используйте Connections.

Переопределение params при запуске зависит от настройки core.dag_run_conf_overrides_params. Если задача использует значение по умолчанию вместо введенного, проверьте эту настройку с администратором кластера. Подробнее — в описании Params.

Повторить упавшую задачуПовторить упавшую задачу

Используйте Clear Task Instance, чтобы повторить задачу в существующем запуске. Сначала изучите логи и устраните причину ошибки, например восстановите доступ к внешнему сервису или исправьте Connection. Убедитесь, что повторная обработка данных допустима.

1. Откройте Dags → нужный DAG → Runs и выберите запуск с ошибкой.

2. На графе или в списке задач выберите упавшую задачу. В ее действиях нажмите Clear Task Instance.

3. В окне подтверждения выберите область повторного выполнения:

   • Downstream — включить зависимые задачи этого запуска. Этот вариант выбран по умолчанию; выключите его, если нужно повторить только выбранную задачу.
   • Upstream — включить предшествующие задачи.
   • Past и Future — включить задачи из других запусков. Для повтора в пределах одного запуска оставьте их выключенными.
   • Clear only failed tasks — ограничить выбор задачами с ошибками.

4. Проверьте список Affected Tasks и нажмите Confirm.

5. Убедитесь, что DAG не приостановлен. Дождитесь повторного выполнения выбранных задач, затем проверьте их состояния и новую попытку на вкладке Logs.

Если для исправления пришлось обновить Python-код, проверьте загруженную версию перед повтором. Доступность выбора версии в окне Clear зависит от способа хранения DAG; само обновление файла не подтверждает, какой код использует повторная попытка.

Присвоение задаче состояния success вручную меняет ее состояние без выполнения кода. Для повторного выполнения используйте Clear Task Instance. Подробнее — в документации по повторному выполнению задач.

Приостановить или возобновить DAGПриостановить или возобновить DAG

1. В разделе Dags найдите нужный DAG.
2. Выключите переключатель рядом с именем, чтобы приостановить DAG, или включите его, чтобы возобновить планирование.

Приостановка DAG не завершает уже выполняющиеся задачи и не останавливает кластер. Следующие задачи ожидают возобновления DAG. Подробнее — в описании приостановки DAG. Для остановки основных компонентов используйте приостановку кластера в консоли управления.

Настроить расписаниеНастроить расписание

Расписание задается в Python-коде DAG. В примере замените schedule=None на schedule="0 9 * * *", сохранив остальные параметры. Этот cron задает ежедневный запуск в 09:00 UTC: часовой пояс определяется start_date, который в примере задан с timezone.utc.

1. Обновите DAG-файл в Git или S3 и проверьте изменение на вкладке Code.
2. Откройте сведения о DAG на вкладке Details и проверьте расписание. Учитывайте часовой пояс отображения дат в интерфейсе.
3. Включите DAG переключателем рядом с именем.
4. После наступления времени по расписанию откройте Runs. Проверьте появление запуска типа scheduled и его результат.

При catchup=False, как в примере, Apache Airflow™ не создает все пропущенные запуски с даты start_date. При catchup=True включение или возобновление DAG может привести к обработке пропущенных интервалов. Значение schedule=None отключает автоматические запуски даже у включенного DAG.

Для других расписаний учитывайте различие между временем запуска и логической датой обработки данных. Подробнее — в описании запусков DAG.

Обработать прошлый период через BackfillОбработать прошлый период через Backfill

Backfill создает запуски для выбранного прошлого периода по расписанию DAG. Для проверки сначала настройте расписание: у исходного hello_airflow с schedule=None Backfill недоступен.

1. Откройте DAG и нажмите Trigger → Backfill.

2. В Date Range задайте начало From и конец To периода. Проверьте часовой пояс дат.

3. В Reprocess Behavior выберите:

   • Missing Runs — создать только отсутствующие запуски;
   • Missing and Errored Runs — также повторно обработать даты запусков с ошибками;
   • All Runs — также обработать даты уже завершенных запусков.

4. В Max Active Runs задайте допустимое число параллельных запусков для этой операции. Для первой проверки используйте 1.

5. При необходимости включите Run Backwards, чтобы начать с более поздних дат. Параметры DAG задайте в Run Parameters или в Advanced Options → Configuration JSON, если они требуются для обработки выбранного периода.

6. Проверьте рассчитанное количество запусков. Если DAG приостановлен, включите его с помощью Unpause … on trigger. Нажмите Run Backfill.

7. Откройте вкладку Backfills, затем проверьте созданные запуски на вкладке Runs и результаты задач.

Перед выбором All Runs убедитесь, что повторная обработка не создаст нежелательные дубликаты во внешних системах. Если для даты уже есть запуск в состоянии running или queued, новый запуск для нее не создается. Подробнее — в описании Backfill.