Mengelompokkan tugas di dalam DAG

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Catatan: Halaman ini belum direvisi untuk Cloud Composer 3 dan menampilkan konten untuk Cloud Composer 2.

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

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

Catatan: Airflow menyediakan SubDAGs untuk mengatasi kasus penggunaan tugas berulang. Meskipun merupakan pola desain umum untuk mengelompokkan tugas, SubDAG sering menyebabkan masalah performa dan fungsional. Sebaiknya hindari penggunaan SubDAG untuk mengelompokkan tugas dalam alur kerja Anda dan pilih salah satu pendekatan alternatif yang dijelaskan dalam halaman ini.

Mengelompokkan tugas di grafik DAG

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

Perhatikan contoh berikut:

Dalam alur kerja ini, tugas op-1 dan op-2 dijalankan 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:

Aliran udara 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 anak dari DAG orang tua

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.

Implementasi alur kerja ini memerlukan dua file DAG terpisah. File DAG yang mengontrol terlihat seperti berikut:

Aliran udara 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 pengendali, akan terlihat seperti berikut:

Aliran udara 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 dapat berfungsi.

Mengelompokkan tugas dengan operator TaskGroup

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

Perhatikan contoh berikut:

Tugas op-1 dan op-2 dikelompokkan bersama dalam satu blok dengan ID taskgroup_1. Implementasi alur kerja ini akan 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