Order of execution in Airflow

Question

I'm reviewing the example code here

There are two "operation" function:

def my_sleeping_function(random_base):
    """This is a function that will run within the DAG execution"""
    time.sleep(random_base)

and:

def print_context(ds, **kwargs):
    pprint(kwargs)
    print(ds)
    return 'Whatever you return gets printed in the logs'

For each run of my_sleeping_function we run print_context?

What I don't understand is the order. This is the graph and the Tree.. the execution order isn't the same:

What happens first? What happens after? Why?

I assume that according to this:

for i in range(5):
    task = PythonOperator(
        task_id='sleep_for_' + str(i),
        python_callable=my_sleeping_function,
        op_kwargs={'random_base': float(i) / 10},
        dag=dag)

    task.set_upstream(run_this)

run_this executes and then task executes but the loop confuses me.

Answer 1

I think your confusion is based around you expecting the graph view and tree view to be two separate visualizing of the same thing. However they are used to visualize different things.

The graph view shows the order the tasks will run in your workflow. In your case print_the_context will run, and once complete sleep_for_0, sleep_for_1, sleep_for_2, sleep_for_3, sleep_for_4 will run be run in parallel (or at least as parallel as your airflow config will allow)

The tree view represents a depth first visualization of the DAG (and the status of each task over time in the squares to the right). That is to say the the first level of nodes in the tree are final tasks in the dag (leaf nodes), where the dag will be considered successfully completed. It branches out for each dependent task that must be run for it to run.

To put it another way, the execution order is the same in both views, it is just being visualized from different directions.

Answer 2

The loop here is just showing you how to dynamically build a DAG. The order in which things get executed depends on what you set as the "upstream" or "downstream" tasks.

The example that you linked to could also be done like the following example. However what if you want to add another 10 tasks? You'd have to do quite a lot of copy/paste coding to achieve the same thing, better to just put it in a loop as in the linked example:

def my_sleeping_function(random_base):
    """This is a function that will run within the DAG execution"""
    time.sleep(random_base)


def print_context(ds, **kwargs):
    pprint(kwargs)
    print(ds)
    return 'Whatever you return gets printed in the logs'


run_this = PythonOperator(
    task_id='print_the_context',
    provide_context=True,
    python_callable=print_context,
    dag=dag)

task_0 = PythonOperator(
    task_id='sleep_for_' + 0,
    python_callable=my_sleeping_function,
    op_kwargs={'random_base': float(0) / 10},
    dag=dag)

task_1 = PythonOperator(
    task_id='sleep_for_' + 1,
    python_callable=my_sleeping_function,
    op_kwargs={'random_base': float(1) / 10},
    dag=dag)

task_2 = PythonOperator(
    task_id='sleep_for_' + 2,
    python_callable=my_sleeping_function,
    op_kwargs={'random_base': float(2) / 10},
    dag=dag)


task_3 = PythonOperator(
    task_id='sleep_for_' + 3,
    python_callable=my_sleeping_function,
    op_kwargs={'random_base': float(3) / 10},
    dag=dag)

task_4 = PythonOperator(
    task_id='sleep_for_' + 4,
    python_callable=my_sleeping_function,
    op_kwargs={'random_base': float(4) / 10},
    dag=dag)


task_0.set_upstream(run_this)
task_1.set_upstream(run_this)
task_2.set_upstream(run_this)
task_3.set_upstream(run_this)
task_4.set_upstream(run_this)