AIRFLOW 2.0 TASK GROUPS CODE

Makes UI easier to use especially for DAGs with many tasks.

Makes it possible to organize DAGs with large number tasks into more mangeable groups.

Offers an alternative to SubDagOperator.

Start > section_1 > some_other_task > section_2 > end With TaskGroup('section_2', tooltip='Tasks for Section 2') as section_2: Some_other_task = DummyOperator(task_id='some-other-task') With TaskGroup('section_1', tooltip='Tasks for Section 1') as section_1: With DAG(dag_id='example_task_group', start_date=days_ago(2)) as dag:

AIRFLOW 2.0 TASK GROUPS CODE

This is the code that builds the above DAG:įrom _operator import DummyOperatorįrom _group import TaskGroupĭummies = When it is deserialized, the task_dict of the DAG is looked up to put the corresponding task into the TaskGroup. The serialized representation of TaskGroup contains only children task_id or group_id. It is serialized/deserialized just like other DAG attributes. TaskGroup is added as a new attribute of DAG. And also replaces the children's edges with edges that goes in/out of the parent group. It removes children nodes from the graph that are collapsed. The collapse_group() function does the opposite. It also looks at the list of edges to determine what edges need to be added to the graph when the group is expanded. The expand_group() function examines the nodes dictionary and find the next level of children nodes that need to be displayed. When user clicks on the empty regeion in the expanded group node, collapse_group() is called. expand_group() is called when a group node is clicked. only top level tasks and TaskGroups are shown. When the page is first opened, only the root TaskGroup is expanded, i.e. Two javascript functions expand_group() and collapse_group() are added. Most of the changes happen at the UI layer in Graph View. It also has some meta data such as tooltip and ui_color of the TaskGroup.

The dict represents the tree structure of nested TaskGroup. The root TaskGroup of the dag is converted to a dict called 'nodes' and passed to the UI. The > and << operators apply the operation over all tasks within the TaskGroup. TaskGroup acts as a collection of BaseOperator. Setting dependencies on TaskGroupĭependency relationships can be defined between a TaskGroup and a BaseOperator as well as between a TaskGroup and another TaskGroup. It is prefixed with its TaskGroup's group_id which itself is prefixed with the group_id of its parent TaskGroup.

The task_id of BaseOperator is changed to a property. If no TaskGroup is explicitly given and no other TaskGroup exists, a task is added to the root TaskGroup of the DAG. Once it is added, the task_id of this task is prefixed with the group_id and cannot be changed. The only way to add a task to a TaskGroup is during the task's construction. When a BaseOperator is constructed with a task_group specified, or if it is constructed within the contextmanager of a TaskGroup, it is added to the TaskGroup. All subsequently created TaskGroup and tasks are added to the root TaskGroup unless they are explicitly given a TaskGroup. When a DAG is first created, its task_group is initialized to a TaskGroup with group_id and parent_group set to None. The children can be either a TaskGroup or a BaseOperator.Įvery DAG has a root TaskGroup. Every TaskGroup has zero or more children. There is thus no change to how the scheduler works. Internally, all tasks remain on the same DAG. The UI then renders the graph based on the grouping information. The TaskGroup is passed to the UI as a dictionary. The task_id of the tasks within the group are pre-fixed with the group_id of the TaskGroup. All tasks created in the context of a TaskGroup are added to the group. This proposal introduces TaskGroup as a utility class. This AIP does not aim to remove SubDagOperator, although we may consider deprecating it in the future if TaskGroup proves more useful and easier to maintain. It can be used as an alternative to SubDagOperator.

The goal of this proposal is to introduce a new concept called TaskGroup. Considerations What change do you propose to make?

Airflow 2.0 task groups