Start learning free

LESSON 58 OF 107 · DATA ENGINEERING: ORCHESTRATION

The Order to Run In

Practice

Your dict says what each task waits for. To run the pipeline, an orchestrator needs one order in which every task comes after all the tasks it waits for. Such an order is called a topological order. There can be more than one: extract_customers may run before or after clean_orders, as long as it runs before build_report. With dozens of tasks, nobody can work the order out by hand each time a task is added, so the orchestrator computes it.

Python's standard library does it for you. graphlib.TopologicalSorter(deps) takes a dict in exactly your shape: a task's name, then the list of what it waits for. Its .static_order() gives the names one by one, in a valid order:

from graphlib import TopologicalSorter

steps = {"publish": ["convert"], "convert": ["download"], "download": []}
print(list(TopologicalSorter(steps).static_order()))
# ['download', 'convert', 'publish']

A pipeline can be written wrong. If task A waits for B and B waits for A, neither can ever start: that's a cycle, a loop. Instead of waiting forever, static_order() raises a CycleError. You catch it with try and except CycleError as error:, and error.args[1] is the list of the tasks in the loop, ending with the task it started from. For {"convert": ["publish"], "publish": ["convert"]} it is ['convert', 'publish', 'convert']. This is why Airflow calls a pipeline a DAG (a directed acyclic graph: arrows, no loops) and refuses to load one with a loop.

Your program prints the run order of the shop's five tasks. Then it makes extract_orders wait for send_email and tries again:

extract_orders
extract_customers
clean_orders
build_report
send_email
Loop: extract_orders -> clean_orders -> build_report -> send_email -> extract_orders

Your task: the import, deps and the two calls are already there. Write the function print_order(deps): inside a try, loop over TopologicalSorter(deps).static_order() and print each name; when a CycleError happens, print "Loop:" and the tasks of error.args[1] joined with " -> ".


Assignment
Follow the Coding Tutorial to write print_order, which prints the tasks in an order to run them, or the loop that makes such an order impossible.


Hint
except goes at the same indent as try. " -> ".join(names) puts an arrow between the names of a list.


STEP 1 OF 6
    Ctrl + Enter to check