После ветвления Airflow пропустил общий шаг: выбираем trigger_rule

DAG выбирает полную или быструю загрузку. Одна ветка успешно заканчивается, вторая закономерно получает skipped. Но общий шаг проверки тоже становится skipped. Причина часто в правиле запуска по умолчанию: all_success требует успеха каждого непосредственного предшественника.

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

Две ветки и один контрольный шаг

Ниже полноценный DAG для Airflow 3.x. Параметр full выбирается при ручном запуске. Ветки печатают режим обработки, а общий шаг отмечает достижение точки сверки. Подключите в этих трёх местах собственные загрузчики и проверку количества.

from datetime import datetime, timezone
from airflow.sdk import dag, task
from airflow.providers.standard.operators.empty import EmptyOperator

@dag(schedule=None, start_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
     catchup=False, params={"full": False})
def branching_load():
    @task.branch
    def choose(params=None):
        return "full_load" if params["full"] else "quick_load"

    @task
    def full_load():
        print("Full input selected")

    @task
    def quick_load():
        print("Incremental input selected")

    join = EmptyOperator(task_id="validated_route",
                         trigger_rule="none_failed_min_one_success")
    decision = choose()
    full = full_load()
    quick = quick_load()
    decision >> [full, quick]
    [full, quick] >> join

branching_load()

При full=false задача quick_load успешно завершится, full_load будет skipped, validated_route выполнится. При full=true роли поменяются. Оператор EmptyOperator нужен здесь только как видимая точка соединения; его успешный статус не проверяет содержимое данных.

Значение, которое возвращает branch, должно совпадать с task_id непосредственно доступной ветки. Если поместить задачи в TaskGroup, учитывайте префикс группы. Ошибка идентификатора относится к выбору ветки и не лечится trigger_rule общего шага.

Почему не all_done

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

none_failed допускает ситуацию, когда все родители пропущены. Если общий шаг требует хотя бы один реальный результат, добавление min_one_success отражает этот смысл. При сложном графе отдельно нарисуйте непосредственных родителей join: правило оценивает их состояния, а не произвольные задачи где-то выше.

Проверим три сценария

Запустите обе успешные альтернативы, затем намеренно вызовите исключение внутри выбранной ветки учебного DAG. В последнем случае validated_route не должен завершиться успешно. Это проверяет, что новое правило разрешает ожидаемый пропуск, но сохраняет блокировку при ошибке.

Также проверьте сценарий, где выбираются несколько веток, если ваш branch это допускает. Один успех не должен скрывать падение другой выбранной ветки. Именно поэтому здесь важно условие отсутствия ошибок, а не только наличие успешного предшественника.

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

В курсе Data Engineering Fast-Track вы отработаете зависимости и расписания Airflow, затем включите их в общий pipeline с Kafka и Spark. Изучите программу и научитесь проектировать граф, где статус задачи соответствует готовности следующего этапа.