Mengelompokkan tugas di dalam DAG

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Halaman ini menjelaskan cara mengelompokkan tugas di pipeline Airflow menggunakan pola desain berikut:

• Mengelompokkan tugas dalam grafik DAG.
• Memicu DAG turunan dari DAG induk.
• Mengelompokkan tugas dengan operator TaskGroup (hanya dengan Airflow 2).

Catatan: Airflow menyediakan SubDAGs untuk menangani kasus penggunaan tugas berulang. Meskipun merupakan pola desain umum untuk mengelompokkan tugas, SubDAG sering kali menyebabkan masalah performa dan fungsi. Anda harus menghindari penggunaan SubDAG untuk mengelompokkan tugas dalam alur kerja dan memilih salah satu pendekatan alternatif yang dijelaskan di halaman ini.

Mengelompokkan tugas dalam grafik DAG

Untuk mengelompokkan tugas dalam fase tertentu pada pipeline, Anda dapat menggunakan hubungan antara tugas dalam file DAG.

Perhatikan contoh berikut:

Dalam alur kerja ini, tugas op-1 dan op-2 berjalan bersama setelah tugas awal start. Anda dapat melakukannya dengan mengelompokkan tugas bersama dengan pernyataan start >> [task_1, task_2].

Kode berikut memberikan implementasi lengkap DAG di atas:

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

Aliran udara 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

Memicu DAG turunan dari DAG induk

Anda dapat memicu satu DAG dari DAG lain dengan TriggerDagRunOperator. Bergantung pada versi Airflow, Anda dapat menemukan operator TriggerDagRunOperator di modul yang berbeda:

• Untuk Airflow 1.10.*, gunakan modul airflow.operators.dagrun_operator.
• Untuk Airflow 2, gunakan modul airflow.operators.trigger_dagrun.

Perhatikan contoh berikut:

Dalam alur kerja ini, blok dag_1 dan dag_2 mewakili serangkaian tugas yang dikelompokkan bersama dalam DAG terpisah di lingkungan Cloud Composer.

Penerapan alur kerja ini memerlukan dua file DAG terpisah. File DAG pengontrol terlihat seperti berikut:

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

Aliran udara 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

Catatan: Nilai untuk trigger_dag_id di dalam TriggerDagRunOperator harus cocok dengan nilai dag_id DAG yang ingin Anda picu.

Implementasi DAG turunan, yang dipicu oleh DAG kontrol, terlihat seperti berikut:

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

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

Anda harus mengupload kedua file DAG di lingkungan Cloud Composer agar alur kerja berfungsi.

Mengelompokkan tugas dengan operator TaskGroup

Di Airflow 2, Anda dapat menggunakan operator TaskGroup untuk mengelompokkan tugas dalam DAG. Tugas yang ditentukan dalam blok TaskGroup masih merupakan bagian dari DAG utama.

Perhatikan contoh berikut:

Tugas op-1 dan op-2 dikelompokkan bersama dalam blok dengan ID taskgroup_1. Penerapan alur kerja ini terlihat seperti kode berikut:

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