DAG 中的分組工作

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本頁面說明如何使用下列設計模式，在 Airflow 管道中將工作分組：

  • 在 DAG 圖表中將工作分組。
  • 從父項 DAG 觸發子項 DAG。
  • 使用 TaskGroup 運算子將工作分組。

在 DAG 圖表中將工作分組

如要在管道的特定階段中將工作分組，可以使用 DAG 檔案中工作之間的關係。

請見如下範例：

顯示分支工作的 Airflow 工作圖
圖 1. 工作可以分組在 Airflow DAG 中 (按一下可放大)

在這個工作流程中，初始工作 start 完成後，工作 op-1 和 op-2 會一併執行。如要達成這個目標，請使用 start >> [task_1, task_2] 陳述式將工作分組。

以下範例提供這個 DAG 的完整實作方式：

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago

DAG_NAME = "all_tasks_in_one_dag"

args = {"owner": "airflow", "start_date": days_ago(1), "schedule_interval": "@once"}

with DAG(dag_id=DAG_NAME, default_args=args) as dag:
    start = DummyOperator(task_id="start")

    task_1 = BashOperator(task_id="op-1", bash_command=":", dag=dag)

    task_2 = BashOperator(task_id="op-2", bash_command=":", dag=dag)

    some_other_task = DummyOperator(task_id="some-other-task")

    task_3 = BashOperator(task_id="op-3", bash_command=":", dag=dag)

    task_4 = BashOperator(task_id="op-4", bash_command=":", dag=dag)

    end = DummyOperator(task_id="end")

    start >> [task_1, task_2] >> some_other_task >> [task_3, task_4] >> end

從父項 DAG 觸發子項 DAG

您可以使用 TriggerDagRunOperator 運算子，從一個 DAG 觸發另一個 DAG。

請見如下範例：

Airflow 工作圖，顯示做為 DAG 圖一部分觸發的子項 DAG
圖 2. 您可以使用 TriggerDagRunOperator 從 DAG 內觸發 DAG (按一下可放大)

在這個工作流程中，dag_1 和 dag_2 區塊代表一系列工作，這些工作會分組在 Cloud Composer 環境的獨立 DAG 中。

如要實作這個工作流程，需要兩個不同的 DAG 檔案。 控制 DAG 檔案如下所示：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.utils.dates import days_ago


with DAG(
    dag_id="controller_dag_to_trigger_other_dags",
    default_args={"owner": "airflow"},
    start_date=days_ago(1),
    schedule_interval="@once",
) as dag:
    start = DummyOperator(task_id="start")

    trigger_1 = TriggerDagRunOperator(
        task_id="dag_1",
        trigger_dag_id="dag-to-trigger",  # Ensure this equals the dag_id of the DAG to trigger
        conf={"message": "Hello World"},
    )
    trigger_2 = TriggerDagRunOperator(
        task_id="dag_2",
        trigger_dag_id="dag-to-trigger",  # Ensure this equals the dag_id of the DAG to trigger
        conf={"message": "Hello World"},
    )

    some_other_task = DummyOperator(task_id="some-other-task")

    end = DummyOperator(task_id="end")

    start >> trigger_1 >> some_other_task >> trigger_2 >> end

由控制 DAG 觸發的子項 DAG 實作方式如下：

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago

DAG_NAME = "dag-to-trigger"

args = {"owner": "airflow", "start_date": days_ago(1), "schedule_interval": "None"}

with DAG(dag_id=DAG_NAME, default_args=args) as dag:
    dag_task = DummyOperator(task_id="dag-task")

您必須在 Cloud Composer 環境中上傳這兩個 DAG 檔案，DAG 才能正常運作。

使用 TaskGroup 運算子將工作分組

您可以使用 TaskGroup 運算子，將 DAG 中的工作分組。TaskGroup 區塊中定義的工作仍屬於主要 DAG。

請見如下範例：

顯示兩個工作群組的 Airflow 工作圖
圖 3. 工作可以在 UI 中以視覺化方式分組，並使用 TaskGroup 運算子 (按一下即可放大)

工作 op-1 和 op-2 會在 ID 為 taskgroup_1 的區塊中分組。這個工作流程的實作方式如下列程式碼所示：

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.dates import days_ago
from airflow.utils.task_group import TaskGroup

with DAG(dag_id="taskgroup_example", start_date=days_ago(1)) as dag:
    start = DummyOperator(task_id="start")

    with TaskGroup("taskgroup_1", tooltip="task group #1") as section_1:
        task_1 = BashOperator(task_id="op-1", bash_command=":")
        task_2 = BashOperator(task_id="op-2", bash_command=":")

    with TaskGroup("taskgroup_2", tooltip="task group #2") as section_2:
        task_3 = BashOperator(task_id="op-3", bash_command=":")
        task_4 = BashOperator(task_id="op-4", bash_command=":")

    some_other_task = DummyOperator(task_id="some-other-task")

    end = DummyOperator(task_id="end")

    start >> section_1 >> some_other_task >> section_2 >> end

後續步驟