Proper way to define Airflow DAG task ordering

Question

I have a rather long set of tasks in a DAG and each has quite a long task_id, the details are all relevant, and the naming can't be shortened.
Currently I have written it like:

a_very_long_long_named_task_1 >> a_very_long_long_named_task_2 >> a_very_long_long_named_task_3 >> a_very_long_long_named_task_4 >> a_very_long_long_named_task_5

In other DAGs, I have seen this be split into multiple lines, albeit with duplication:

a_very_long_long_named_task_1 >> a_very_long_long_named_task_2
a_very_long_long_named_task_2 >> a_very_long_long_named_task_3
a_very_long_long_named_task_3 >> a_very_long_long_named_task_4
a_very_long_long_named_task_4 >> a_very_long_long_named_task_5

Which is recommended? Is there a best practice, or perhaps another better way to define task ordering?

y2k-shubham y2k-shubham · Accepted Answer · 2020-09-07T11:15:18

You can keep adding your tasks to python list (or dict / something similar) as and when you create (instantiate) them
Then at the end you can wire them up programmatically

Note that the snippet is untested

from typing import List
from airflow.models.baseoperator import BaseOperator

my_tasks: List[BaseOperator] = [
    a_very_long_long_named_task_1,
    a_very_long_long_named_task_2,
    a_very_long_long_named_task_3, 
    a_very_long_long_named_task_4, 
    a_very_long_long_named_task_5
]

..

# define a utility method to set dependencies b/w tasks
def wire_tasks(my_tasks: List[BaseOperator]) -> None:
    """
    A utility method that accepts a list of tasks and links them up
    :param my_tasks: List of tasks (operator instances)
    :type my_tasks: List[BaseOperator]
    :return None
    """
    for i in range(1, len(my_tasks)):
        # this is equivalent to my_tasks[i - 1].set_upstream(my_tasks[i])
        my_tasks[i - 1] >> my_tasks[i]

# call the utility method to wire the tasks
wire_tasks(my_tasks=my_tasks)

Proper way to define Airflow DAG task ordering

1 Answers