Branching, Trigger Rules & Task Groups

A branch lets a run choose a path. Every task stays in the graph, but only some of them run.

The tasks that do not run become skipped. And a skip can spread to tasks you wanted to run.

This chapter shows:

1. How far a skip travels, and how a trigger rule stops it.
2. The trigger rules you will really use.
3. When to branch, when to short-circuit, and when to use a plain if.
4. Task groups, depends_on_past, and setup and teardown tasks.

Here is a branch. It picks full_load or incremental_load. Both paths then meet at a join:

@task.branch
def choose():
    return "full_load"          # the task_id to run

full = EmptyOperator(task_id="full_load")
inc = EmptyOperator(task_id="incremental_load")
join_default = EmptyOperator(task_id="join_default")
after_default = EmptyOperator(task_id="after_default")
join_fixed = EmptyOperator(task_id="join_fixed",
                           trigger_rule="none_failed_min_one_success")

choose() >> [full, inc] >> join_default >> after_default
[full, inc] >> join_fixed

A trigger rule decides when a task may start, from the states of the tasks before it. The default rule is all_success: every upstream task must succeed.

incremental_load was skipped, so join_default was skipped too. Then after_default was skipped, and so on, to the end of the DAG.

No task failed, so the run is green and no alert fires. But the dashboard was never refreshed.

Airflow 3.1 has 13 trigger rules. Four cover almost every case:

Read each name word by word, and it tells you the rule. none_failed_min_one_success means: none failed, minimum one success.

Two similar rules differ when everything upstream was skipped. We ran a branch that chose no path:

So for a join, none_failed_min_one_success is the safer choice. It does not run on an empty pipeline.

There are three ways to make a DAG depend on a condition:

A condition belongs in the graph only when it changes which tasks run. If it only changes what one task does, put an if inside that task.

A branch returns task_ids as text, so Python cannot check them. In our test, a typo stopped the branch task:

@task.branch
def choose():
    return "ful_load"     # should be full_load

Keep task_ids in constants, such as FULL = "full_load", and use them in both places.

A task group puts related tasks in one box in the Graph view. It only changes how the graph looks, and how task_ids are named:

with TaskGroup("load") as load:
    EmptyOperator(task_id="orders")
    EmptyOperator(task_id="customers")

start >> load >> build_daily_revenue

The group name is added in front of each task_id. Remember this when you search logs, or when a branch returns a task_id inside a group.

Old Airflow had SubDAGs: a DAG inside a task. They caused deadlocks and were hard to use, and Airflow 3.0 removed them. Use task groups instead.

With depends_on_past=True, a task waits for its own instance in the previous run to succeed. Use it when each day builds on the day before, such as a running total.

The cost is real. If one day fails, every later day waits until someone fixes it.

Setup and teardown tasks solve a different problem. Suppose a task creates a cluster, and a later task must delete it, even if the work fails:

@task
def create_cluster():
    print("cluster created")

@task
def work():
    raise ValueError("work failed")

@task
def delete_cluster():
    print("cluster deleted")

cc = create_cluster()
cc >> work() >> delete_cluster().as_teardown(setups=cc)

With a normal task, delete_cluster would be upstream_failed, and the cluster would keep running and costing money.

Why not an all_done task? In our test, work failed, and an all_done clean-up after it succeeded. The whole run was then marked successful.

A teardown does not hide the failure like that. Also, clearing the work task clears its setup and teardown too.

Branch, rejoin, clean up:

1. A skipped task skips everything after it under the default rule.
2. Give the join after a branch none_failed_min_one_success.
3. Use all_done only for clean-up, never to read a result.
4. Branch for which path, short-circuit for any path, and a plain if for what one task does.
5. Use a teardown task to delete what a setup task created.

Check yourself: after a branch, the run is green but the dashboard never refreshed. Where do you look? (Answer: the trigger rule on the join.)