气流任务流 - 并行运行任务 [英] Airflow taskflow - run task in parallele

问题描述

想要尝试新的任务流 API，我到了需要 2 个并行任务的地步.

Wanted to try the new taskflow API I came to the point where I need to have 2 parallels task.

使用 Airflow v1，我曾经做过类似的事情

With Airflow v1 I was use to do something like

task_1 >> [task_2, task_3]
[task_2, task_3] >> task_4

对于PythonOperator

我如何使用 TaskFlow 做列表?

How can I do the list with TaskFlow ?

谢谢

推荐答案

如果每个任务都依赖于上一个任务的值，您可以通过以下方式实现:

if each task is depended on the value from previous task you can achieve it by:

from airflow.utils.dates import days_ago
from airflow.decorators import task, dag


@task
def task_1():
    return 'first task'

@task
def task_2(value):
    return 'second task'

@task
def task_3(value):
    return 'third task'

@task
def task_4(value1, value2):
    return 'forth task'

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(2),
}


@dag(dag_id='taskflow_stackoverflow', schedule_interval='@once', default_args=default_args, catchup=False)
def my_dag():
    op_1 = task_1()
    op_2 = task_2(op_1)
    op_3 = task_3(op_1)
    op_4 = task_4(op_2, op_3)

dag = my_dag()

您提到的语法也受支持，但您无法直接访问先前任务中的 xcom 值:

The syntax that you mentioned is also supported but you won't get direct access to the xcom values from previous tasks:

@task
def task_1():
    return 'first task'

@task
def task_2():
    return 'second task'

@task
def task_3():
    return 'third task'

@task
def task_4():
    return 'forth task'

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(2),
}


@dag(dag_id='taskflow_stackoverflow', schedule_interval='@once', default_args=default_args, catchup=False)
def my_dag():

    op_1 = task_1()
    op_2 = task_2()
    op_3 = task_3()
    op_4 = task_4()

    op_1 >> [op_2, op_3]
    [op_2, op_3] >> op_4

dag = my_dag()

可能您需要根据您想要实现的目标混合使用两种语法选项.

Probably you need to mix the two options of syntax depending on what you want to achieve.

这篇关于气流任务流 - 并行运行任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！