Airflow trigger_rule all_done not working as expected

user_18492837 0
Viewed 226

I have the following DAG in Airflow 1.10.9 where the clean_folder task should run once all the previous tasks either succeeded, failed or were skipped. To ensure this, I put the trigger_rule parameter of the clean_folder operator to "all_done":

t_clean_folders = BashOperator
            bash_command=f"python {os.path.join(custom_resources_path, 'cleaning.py')} {args['n_branches']}",
            trigger_rule='all_done',
            task_id="clean_folder",
        )

This logic works properly when all tasked in the branches executed before are skipped: [Graph view][1]

However when a branch was successfully executed, the clean_folder task is skipped: [Graph view][2]

The branches are defined dynamically as follow:

for b in range(args['n_branches']):
        t_file_sensing = FileSensor(
            filepath=f"{input_path}/input_{b}",
            task_id=f"file_sensing_{b}",
            poke_interval=60,
            timeout=60*60*5,
            soft_fail=True,
            retries=3,
        )
        t_data_staging = BashOperator(
            bash_command=f"python {os.path.join(custom_resources_path, 'staging.py')} {b}",
            task_id=f"data_staging_{b}",
        )
        ...

The documentation provides the following definition of "all_done": all parents are done with their execution. Is this normal behavior of the trigger_rule? What can I change to ensure clean_folder will run in any case (and last)? Thanks! [1]: https://i.stack.imgur.com/AGqqD.png [2]: https://i.stack.imgur.com/7CG3r.png

airflow
1 Answers

If possible, you should consider to upgrade your Airflow version to at least 1.10.15 in order to benefit from more recent bug-fixes.

It really surprises me that clean_folder and dag_complete both get executed when every parent tasks are skipped. The behaviour when a task is skipped is to directly skip its child tasks without first checking their trigger_rules.

According to Airflow 1.10.9 Documentation on trigger_rules,

Skipped tasks will cascade through trigger rules all_success and all_failed but not all_done [...]

For your UseCase, you could split the workflow into 2 DAGs:

  • 1 DAG to do everything you want except the t_clean_folder
  • 1 DAG to execute the t_clean_folder task, preceded by an ExternalTaskSensor
user_25084680
Related