Raggruppamento di attività all'interno dei DAG

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Nota: questa pagina non è ancora stata rivista per Cloud Composer 3 e mostra contenuti per Cloud Composer 2.

Questa pagina descrive come raggruppare le attività nelle pipeline Airflow utilizzando i seguenti pattern di progettazione:

• Raggruppamento delle attività nel grafico DAG.
• Attivazione dei DAG secondari da un DAG principale.
• Raggruppamento delle attività con l'operatore TaskGroup (solo con Airflow 2).

di Gemini Advanced.

Nota: Airflow fornisce SubDAGs per affrontare il caso d'uso delle attività ricorrenti. Nonostante sia un modello di progettazione comune per raggruppare le attività, I subDAG spesso causano problemi di prestazioni e funzionalità. Ti consigliamo di evitare di utilizzare i DAG secondari per raggruppare le attività nel tuo flusso di lavoro e di preferire uno degli approcci alternativi descritti in questa pagina.

Raggruppamento delle attività nel grafico DAG

Per raggruppare le attività in determinate fasi della pipeline, puoi utilizzare le relazioni tra le attività nel file DAG.

Considera l'esempio seguente:

In questo flusso di lavoro, le attività op-1 e op-2 vengono eseguite insieme dopo il l'attività start. Puoi farlo raggruppando le attività con l'istruzione start >> [task_1, task_2].

Il seguente codice fornisce un'implementazione completa del DAG riportato sopra:

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

Attivazione dei DAG secondari da un DAG principale

Puoi attivare un DAG da un altro DAG con il TriggerDagRunOperator. A seconda della versione di Airflow, puoi trovare l'operatore TriggerDagRunOperator in un altro modulo:

• Per Airflow 1.10.*, utilizza il airflow.operators.dagrun_operator in maggior dettaglio più avanti in questo modulo.
• Per Airflow 2, utilizza airflow.operators.trigger_dagrun in maggior dettaglio più avanti in questo modulo.

Considera l'esempio seguente:

In questo flusso di lavoro, i blocchi dag_1 e dag_2 rappresentano una serie di attività raggruppati in un DAG separato in Cloud Composer completamente gestito di Google Cloud.

L'implementazione di questo flusso di lavoro richiede due file DAG separati. Il file DAG di controllo ha il seguente aspetto:

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

Nota: il valore di trigger_dag_id all'interno di TriggerDagRunOperator deve corrispondere al valore dag_id del DAG che vuoi attivare.

L'implementazione del DAG secondario, attivato dal DAG di controllo, è la seguente:

Flusso d'aria 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")

Affinché il flusso di lavoro funzioni, devi caricare entrambi i file DAG nel tuo ambiente Cloud Composer.

Raggruppamento delle attività con l'operatore TaskGroup

In Airflow 2, puoi utilizzare Operatore TaskGroup per raggruppare le attività nel tuo DAG. Le attività definite all'interno di un blocco TaskGroup fanno ancora parte del DAG principale.

Considera l'esempio seguente:

Le attività op-1 e op-2 sono raggruppate in un blocco con ID taskgroup_1. L'implementazione di questo flusso di lavoro è simile al seguente codice:

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