Commands
Apache Airflow

airflow tasks test: Тестовый запуск задачи без записи в БД: отладка без side-эфф

Тестовый запуск задачи без записи в БД: отладка без side-эффектов

#command#airflow#airflow

Базовый синтаксис

Команда airflow tasks test — это ваш главный инструмент для отладки в Airflow. Она запускает конкретную задачу (task) в изоляции, без регистрации статусов в метабазе Airflow и без отправки зависимостей на выполнение. Это означает, что ни task_instance, ни dag_run в базе не создаются, и триггерные правила (all_success, all_failed) не работают.

airflow tasks test my_dag_id my_task_id 2024-01-01

Разберем по частям:

  • my_dag_id — идентификатор вашего DAG.
  • my_task_id — идентификатор конкретной задачи внутри этого DAG.
  • 2024-01-01 — дата выполнения (execution_date). Это ключевой параметр, который подставляется в шаблоны Jinja и макросы типа {{ ds }}. Формат — YYYY-MM-DD.

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

Полезные флаги

Флаг --subdir критически важен при разработке. Airflow ищет DAG-файлы по пути, указанному в dags_folder в airflow.cfg. Если вы запускаете команду не из той же директории, где лежит ваш Dаг, или у вас несколько папок с DAG, Airflow его не найдет.

airflow tasks test my_dag_id my_task_id 2024-01-01 --subdir /home/user/dags/my_dag_file.py

Этот флаг явно указывает Airflow, в каком конкретно файле искать определение DAG. Это избавляет от ошибки Dag id ... not found.

Типичные сценарии

1. Отладка сложной логики в PythonOperator. Вы написали кастомную функцию, которая парсит JSON из API, и она падает с непонятной ошибкой. Вместо того чтобы запускать весь DAG и ждать, пока дойдет до вашей задачи, тестируете ее сразу:

airflow tasks test data_processing_dag transform_api_response_task 2024-10-28

В логах терминала вы сразу увидите полный stack trace и сможете быстро исправить код, например, обработав случай с None.

2. Проверка подстановки переменных и работы с XCom. Нужно убедиться, что шаблонизация {{ ds }} или {{ data_interval_start }} работает как ожидается, и что upstream-таска корректно кладет значение в XCom, которое читает downstream-таска.

# Сначала тестируем таску, которая пушит данные
airflow tasks test my_dag push_data_task 2024-10-28
# Смотрим вывод, убеждаемся, что значение есть.
# Затем тестируем таску, которая тянет это значение
airflow tasks test my_dag pull_data_task 2024-10-28

Вы увидите, какое значение было фактически получено через ti.xcom_pull().

Частые ошибки

1. Неверный формат execution_date. Самая частая ошибка — передать дату в неправильном формате. Команда ожидает дату в формате YYYY-MM-DD. Если вы укажете 2024/10/28 или 28-10-2024, получите ошибку. Для даты-времени используйте полный ISO-формат: 2024-10-28T14:30:00.

2. Ожидание side-эффектов, которых не будет. Запомните: tasks test не создает записей в БД Airflow. Это означает, что если ваша задача рассчитана на взаимодействие с TaskInstance (например, чтение специального поля из базы) или использует сенсоры — они не будут работать как в продакшене. Сенсор в тестовом режиме просто вернет True. Это инструмент для отладки логики, а не полного запуска.