Airflow 任务流 - 并行运行任务

Question

想要尝试新的任务流 API，我需要有 2 个并行任务。

使用 Airflow v1，我习惯做类似的事情

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

现在我们调用任务的方式有所不同PythonOperator

如何使用 TaskFlow 制作列表？

谢谢

Answer 1

如果每个任务都取决于前一个任务的价值，您可以通过以下方式实现：

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 值：

@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()

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