python - Airflow 中的执行顺序

Question

我正在查看示例代码 here

有两个“操作”函数:

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'

对于 my_sleeping_function 的每次运行，我们都会运行 print_context?

我不明白的是顺序。这是图和树..执行顺序不一样:

首先会发生什么？之后会发生什么？为什么？

我假设根据这个:

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 执行，然后任务执行，但循环让我感到困惑。

Answer 1

我认为您的困惑是基于您期望图形 View 和 TreeView 是同一事物的两个独立可视化。但是，它们用于可视化不同的事物。

图 TableView 显示了任务将在您的工作流程中运行的顺序。在你的情况下 print_the_context 将运行，一旦完成 sleep_for_0, sleep_for_1, sleep_for_2, sleep_for_3, sleep_for_4 将并行运行(或至少与您的 Airflow 配置允许的并行运行)

树形 View 表示 DAG 的深度优先可视化(以及右侧方 block 中每个任务随时间推移的状态)。也就是说树中的第一级节点是dag中的最终任务(叶子节点)，在这里dag将被认为成功完成。它为每个必须运行的依赖任务分支出来。

换句话说，两个 View 的执行顺序是相同的，只是从不同的方向进行可视化。