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.

Mengelompokkan tugas dalam grafik DAG

Untuk mengelompokkan tugas dalam fase tertentu di pipeline, Anda dapat menggunakan hubungan antar-tugas dalam file DAG.

Perhatikan contoh berikut:

Grafik tugas Airflow yang menampilkan tugas percabangan
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 menggunakan pernyataan start >> [task_1, task_2].

Contoh berikut memberikan penerapan 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:

Grafik tugas Airflow yang menampilkan DAG turunan yang dipicu sebagai bagian dari grafik DAG
Gambar 2. DAG dapat dipicu dari dalam DAG dengan TriggerDagRunOperator (klik untuk memperbesar)

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

Penerapan alur kerja ini memerlukan dua file DAG terpisah. File DAG yang mengontrol akan 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

Implementasi DAG turunan, yang dipicu oleh DAG pengontrol, akan 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 tetap menjadi bagian dari DAG utama.

Perhatikan contoh berikut:

Grafik tugas Airflow yang menampilkan dua grup tugas
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