Below is a minimal implementation of a branch operator using the taskflow api. The dag will execute either odd_task or even_task based on the string given by branch_on_condition. odd_task or even_task will also use the return_int value on which the branching decision was made. Then the final_task will be executed. All straight forward. from airflow.decorators import dag, task @dag(dag_id='Example_Dag_Simple_Branch') def simple_dag(): @task(task_id='return_int') def return_int(): return 3 @task.branch(task_id='branch_on_condition') def branch_on_condition(upstream_value): if upstream_value & 1: return 'odd_task' else: return 'even_task' @task(task_id='odd_task') def odd_task(input_val): print(f"{input_val} is an odd number") return input_val @task(task_id='even_task') def even_task(input_val): print(f"{input_val} is an even number") return input_val @task(task_id='final_task', trigger_rule='one_success') def final_task(): print('final task executed') return returned_int = return_int() branch_value = branch_on_condition(upstream_value=returned_int) even_task_return = even_task(input_val=returned_int) odd_task_return = odd_task(input_val=returned_int) final_return = final_task() branch_value >> [even_task_return, odd_task_return] >> final_return simple_dag() This logs either INFO - 3 is an odd number or INFO - 2 is an even number. Why am I unable to implement a similar pattern using dynamically mapped tasks? If the return_int task were to now return an n length list of integers return_list I can dynamically branch for this list fine, but BOTH branches are executed! from airflow.decorators import dag, task @dag(dag_id='Example_Dag_Dynamic_Branch') def simple_dag(): @task(task_id='return_list') def return_list(): return [1,2,3,4,5,6,7] @task.branch(task_id='branch_on_condition') def branch_on_condition(upstream_value): if upstream_value & 1: return 'odd_task' else: return 'even_task' @task(task_id='odd_task') def odd_task(input_val): print(f"{input_val} is an odd number") return input_val @task(task_id='even_task') def even_task(input_val): print(f"{input_val} is an even number") return input_val @task(task_id='final_task', trigger_rule='one_success') def final_task(): print('final task executed') return returned_list = return_list() branch_value = branch_on_condition.expand(upstream_value=returned_list) even_task_return = even_task.expand(input_val=returned_list) odd_task_return = odd_task.expand(input_val=returned_list) final_return = final_task() branch_value >> [even_task_return, odd_task_return] >> final_return simple_dag() And so the output of the odd_task is for example 7 tasks all that say INFO - n is an odd number for n 1 through 7, which is of course not true. I have tried Packing the branch and downstream tasks into a task group. Messing around with .expand() and .partial() Reading related questions: 1. 2. 3. None of these really answers the question! Finally, I am aware that this behavior could literally just a be standalone if else block within an operator. But in reality the if else behavior could become quite complex and might be nice to separate into multiple tasks. Continue reading...
