How to schedule dynamic task groups in Airflow

Pantonaut 0
Viewed 33

I am trying to accomplish the following based on Airflow 2.3 (referencing below pseudo code):

  • I am reading a list of items using an Operator (operator_1)
  • For every item of the list I want to schedule a task group of two Operators (operator_2_1 and operator_2_2) that are supposed to run one after another
  • If there is more than one item in the list, then only one task group must be executed at a time (no parallel execution)
with DAG(
    ...
) as dag:

    # Must be executed once
    operator_1 = SomeOperator(...) # returns a list of items

    # Must be executed "once per item" and only one at a time
    task_group(items)
        operator_2_1 = SomeOtherOperator(...)
        operator_2_2 = SomeOtherOperator(...)

task_group.expand(operator_1)

What I tried so far:

  • Iterate over the list of items and schedule a task group per item (without using dynamic tasks): Works, but the unwanted parallel execution is a problem
  • Using dynamic tasks, which seems to work only for tasks (but not for task groups)

I would appreciate any input on this!

Thank you in advance!

airflow
1 Answers

  1. Using dynamic tasks:
    Mapping a task group is not possible yet, but this is a feature that will be available soon in the next versions.

  2. Control the parallelism of your task groups:
    You can create a new pool task_groups_pool with 1 slot, and use it for the tasks of the task groups, in this case you will not have more than one task of all the task groups running at the same time.
    And to make sure that the task operator_2_2 will be executed after operator_2_1 of the same group and not a task operator_2_1 in another task group, you can use set the priority of operator_2_2 = priority of operator_2_1 + 1, or by using upstream as a weight_rule for the task groups tasks. (Here you can found more info about priority weight)

Hussein Awala0
Related