Airflow 中的执行顺序

Question

我正在这里查看示例代码

有两个“操作”功能：

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

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

图形视图显示任务在您的工作流中运行的顺序。在您的情况下print_the_context将运行，一旦完成sleep_for_0，sleep_for_1, sleep_for_2, sleep_for_3,sleep_for_4将并行运行（或至少与您的气流配置允许的并行）

树视图表示 DAG 的深度优先可视化（以及每个任务在右侧方块中随时间推移的状态）。也就是说树中的第一层节点是dag中的最终任务（叶子节点），在那里dag会被认为是成功完成的。它为每个必须运行的相关任务分支出来才能运行。

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