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 untuk Cloud Composer 2.

Halaman ini menjelaskan cara mengelompokkan tugas di pipeline Airflow Anda 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 yang berulang. Meskipun merupakan pola desain umum untuk mengelompokkan tugas bersama, SubDAG sering menyebabkan masalah performa dan fungsi. Sebaiknya hindari penggunaan SubDAG untuk mengelompokkan tugas bersama dalam alur kerja Anda dan lebih memilih salah satu pendekatan alternatif yang dijelaskan di halaman ini.

Mengelompokkan tugas di grafik DAG

Untuk mengelompokkan tugas dalam fase tertentu dari pipeline, Anda dapat menggunakan relasi di antara tugas di file DAG Anda.

Perhatikan contoh berikut:

Dalam alur kerja ini, tugas op-1 dan op-2 dijalankan bersama setelah tugas 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 airflow.operators.dagrun_operator ruang lingkup modul ini.
• Untuk Airflow 2, gunakan airflow.operators.trigger_dagrun ruang lingkup modul ini.

Perhatikan contoh berikut:

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

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

Mengelompokkan tugas dengan operator TaskGroup

Di Airflow 2, Anda dapat menggunakan Operator TaskGroup untuk mengelompokkan tugas 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