Mengelompokkan tugas di dalam DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

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.

Penting: Airflow menyediakan SubDAG untuk menangani tugas berulang. Meskipun merupakan pola desain umum untuk mengelompokkan tugas, SubDAG sering kali menyebabkan masalah performa dan fungsi, serta tidak digunakan lagi di Airflow. Sebaiknya hindari penggunaan SubDAG untuk mengelompokkan tugas dalam alur kerja Anda dan pilih 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:

Gambar 1. Tugas dapat dikelompokkan dalam DAG Airflow (klik untuk memperbesar)

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].

Contoh berikut memberikan implementasi lengkap DAG ini:

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 operator TriggerDagRunOperator.

Perhatikan contoh berikut:

Gambar 2. DAG dapat dipicu dari dalam DAG dengan TriggerDagRunOperator (klik untuk memperbesar)

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 DAG berfungsi.

Mengelompokkan tugas dengan operator TaskGroup

Pendekatan ini hanya berfungsi di Airflow 2.

Anda dapat menggunakan operator TaskGroup untuk mengelompokkan tugas bersama-sama dalam DAG. Tugas yang ditentukan dalam blok TaskGroup masih merupakan bagian dari DAG utama.

Perhatikan contoh berikut:

Gambar 3. Tugas dapat dikelompokkan secara visual di UI dengan operator TaskGroup (klik untuk memperbesar)

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

Langkah berikutnya

• Menulis DAG
• Menguji DAG