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.

Wichtig :Airflow bietet SubDAGs für wiederkehrende Aufgaben. Trotz eines gängigen Designmusters zum Gruppieren von Aufgaben verursachen SubDAGs oft Leistungsprobleme und funktionale Probleme. Sie sind in Airflow veraltet. Wir empfehlen, keine SubDAGs zu verwenden, um Aufgaben im Workflow zu gruppieren, und einen der auf dieser Seite beschriebenen alternativen Ansätze zu bevorzugen.

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:

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:

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

Hinweis: Der Wert für trigger_dag_id in TriggerDagRunOperator muss mit dem dag_id-Wert des DAG übereinstimmen, den Sie auslösen möchten.

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:

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

• DAGs schreiben
• DAGs testen