Aufgaben in DAGs gruppieren

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Auf dieser Seite wird beschrieben, wie Sie Aufgaben in Ihren Airflow-Pipelines mithilfe der folgenden Designmuster gruppieren können:

  • Gruppieren von Aufgaben in der DAG-Grafik
  • Untergeordnete DAGs aus einem übergeordneten DAG auslösen.
  • Aufgaben mit dem Operator TaskGroup gruppieren.

Aufgaben in der DAG-Grafik gruppieren

Zum Gruppieren von Aufgaben in bestimmten Phasen Ihrer Pipeline können Sie Beziehungen zwischen den Aufgaben in Ihrer DAG-Datei verwenden.

Dazu ein Beispiel:

Diagramm der Airflow-Aufgaben mit Verzweigungsaufgaben
Abbildung 1. Aufgaben können in einem Airflow-DAG zusammengefasst werden (zum Vergrößern klicken)

In diesem Workflow werden die Aufgaben op-1 und op-2 nach der ersten Aufgabe start ausgeführt. Dies erreichen Sie, indem Sie Aufgaben zusammen mit der Anweisung start >> [task_1, task_2] gruppieren.

Das folgende Beispiel enthält eine vollständige Implementierung dieses DAG:

Airflow 2 

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

Airflow 1 


from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dummy_operator 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

Untergeordnete DAGs aus einem übergeordneten DAG auslösen

Sie können mit dem TriggerDagRunOperator-Operator einen DAG aus einem anderen DAG auslösen.

Dazu ein Beispiel:

Der Graph der Airflow-Aufgaben, in dem untergeordnete DAGs dargestellt werden, die als Teil eines DAG-Graphen ausgelöst wurden
Abbildung 2. DAGs können innerhalb eines DAG mit dem TriggerDagRunOperator ausgelöst werden (zum Vergrößern klicken).

In diesem Workflow stellen die Blöcke dag_1 und dag_2 eine Reihe von Aufgaben dar, die in einem separaten DAG in der Cloud Composer-Umgebung gruppiert sind.

Die Implementierung dieses Workflows erfordert zwei separate DAG-Dateien. Die Steuerungs-DAG-Datei sieht so aus:

Airflow 2 

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

Airflow 1 

from airflow import DAG
from airflow.operators.dagrun_operator import TriggerDagRunOperator
from airflow.operators.dummy_operator import DummyOperator
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

Die Implementierung des untergeordneten DAG, die vom Steuerungs-DAG ausgelöst wird, sieht so aus:

Airflow 2 

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")

Airflow 1 

from airflow import DAG
from airflow.operators.dummy_operator 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")

Damit der DAG funktioniert, müssen Sie beide DAG-Dateien in Ihre Cloud Composer-Umgebung hochladen.

Aufgaben mit dem TaskGroup-Operator gruppieren

Dieser Ansatz funktioniert nur in Airflow 2.

Sie können den TaskGroup-Operator verwenden, um Aufgaben in Ihrem DAG zu gruppieren. Aufgaben, die in einem TaskGroup-Block definiert sind, sind weiterhin Teil des Haupt-DAG.

Dazu ein Beispiel:

Diagramm mit Airflow-Aufgaben, das zwei Aufgabengruppen zeigt
Abbildung 3. Aufgaben können in der Benutzeroberfläche mit dem TaskGroup-Operator visuell gruppiert werden (zum Vergrößern klicken).

Die Aufgaben op-1 und op-2 werden in einem Block mit der ID taskgroup_1 gruppiert. Eine Implementierung dieses Workflows sieht so aus:

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

Nächste Schritte