Airflow: Branching and Trigger Rules

Branching in Airflow

The branch operator is a Python function that returns the ID of the next task. Whatever the function returns determines which path the DAG takes. If a path is skipped, it does not mean the task failed. On another run, the condition could be different.

Trigger Rules

The rules include:

• all_failed: all upstream tasks must fail (used for cleanups or alerts)
• all_done: all upstream tasks are finished regardless of status
• one_failed: triggers an alert when at least one thing goes wrong
• one_success: at least one upstream task succeeded (used for tests)
• none_failed: everything except failed status (retry, skip, or success)
• none_failed_min_one_success: the most appropriate rule for branching
• none_skipped: no task can be skipped

Building a Branching DAG

Create a new file branching.py in your DAG folder. Import the necessary modules:

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from datetime import datetime

This function checks the current day of the week and returns the appropriate task ID:

def choose_path():
    from datetime import date
    weekday = date.today().weekday()
    if weekday < 5:
        return 'weekday_task'
    else:
        return 'weekend_task'

Define the tasks that will execute based on the branch:

def weekday_report():
    print("Running full daily report")

def weekend_report():
    print("Running weekend report")

def summarize():
    print("Branch is complete")

Create the DAG block:

with DAG(
    dag_id='branching_dag',
    start_date=datetime(2023, 1, 1),
    schedule=None,
    catchup=False
) as dag:

Define the tasks using the operators:

    choose_path_task = BranchPythonOperator(
        task_id='choose_path',
        python_callable=choose_path
    )

    weekday_task = PythonOperator(
        task_id='weekday_task',
        python_callable=weekday_report
    )

    weekend_task = PythonOperator(
        task_id='weekend_task',
        python_callable=weekend_report
    )

    summary_task = PythonOperator(
        task_id='summary',
        python_callable=summarize,
        trigger_rule='none_failed_min_one_success'
    )

Set the dependencies:

    choose_path_task >> [weekday_task, weekend_task] >> summary_task

Running the DAG

Save the file and go to the Airflow UI. Find the branching_dag in the DAGs section. Trigger a single run. If today is a weekday, the weekday_task runs and the weekend_task is skipped. Check the logs for the weekday_task to see the print statement "Running full daily report". The summary_task will show "Branch is complete".