用气流/云合成器在DAG中建模ParallelFor

Question

我想创建一个DAG，以便在中运行。工作流包含一个ParallelFor，我不知道如何建模。

工作流程如下所示：

task1 >> task2 >> task3 >> task4

其中，task2将数据拆分为x数组。现在，我希望对这些x数组并行运行task3。Task3输出某样东西，task4组合输出。

(您可以在这里找到工作流的图片：https://github.com/Apollo-Workflows/Sentiment-Analysis)

现在，我有两个可能的想法：

1. 有一个简单的语法(就像用于顺序执行的>> )。但我没有发现这样的语法
2. 与子DAG一起工作。我的想法是附加task2，以便创建x个子was (每个数组一个)。subDAG基本上是task3。在完成所有子are之后，它们的输出将被转发到task4。这有可能吗？如果是，我该怎么做？

Answer 1

我已经找到了解决我的问题的办法。它遵循了我的第一个可能的解决方案。只需使用此链接中的机制：

Airflow rerun a single task multiple times on success

Answer 2

我相信，您提到的post是一个可能的想法，指出了在上一个任务结束后如何运行任务的方向。

要并行运行dags，您应该遵循类似于以下结构的结构

from datetime import datetime

from airflow import DAG
from airflow.operators.bash_operator import BashOperator

dag = DAG("dag_paralel", description="Starting tutorial", schedule_interval=None,
          start_date=datetime(2019, 1, 1),
          catchup=False)

task_1 = BashOperator(task_id='task_1', bash_command='echo "This is task 1!"',dag=dag)
task_2 = BashOperator(task_id='task_2', bash_command='echo "This is task 2!"',dag=dag)

task_list = []
max_attempt = 3
for attempt in range(max_attempt):
    data_pull = BashOperator(
        task_id='task_3_{}'.format(attempt),
        bash_command='echo "This is task - 3_{}!"'.format(attempt),
        dag=dag
    )
    task_list.append(data_pull)


data_validation = BashOperator(task_id='task_final', bash_command='echo "We are at the end"',dag=dag)


task_1 >> task_2 >> task_list

task_list >> data_validation

这是用这种方法得到的DAG结构。

​

​