Task group

DAG yang berisi tiga task mudah dibaca pada Graph view, tetapi graf yang tumbuh menjadi puluhan task cepat memenuhi layar dan kehilangan polanya. Task group mengatasi hal itu dengan mengelompokkan beberapa task menjadi satu kotak visual tanpa mengubah perilaku eksekusi DAG. Topik ini membahas decorator @task_group, penamaan task dengan prefix group_id, alternatif pembuatan grup melalui kelas TaskGroup, serta bentuk visual grup pada Graph view. Contoh yang digunakan adalah DAG arsip cadangan dengan tahap pemuatan dan tahap validasi.

Menata task dengan decorator @task_group#

Decorator @task_group mengubah sebuah fungsi menjadi grup bagi task-task yang didefinisikan di dalam tubuh fungsinya. Fungsi grup dipanggil di dalam generator DAG sebagaimana fungsi task, dan pemanggilan itulah yang mendaftarkan grup beserta seluruh task anggotanya pada DAG. Task di dalam grup tetap berada pada DAG yang sama dan mengikuti seluruh pengaturan DAG, sehingga grup hanya mengubah tampilan graf, bukan urutan eksekusinya.

Berkas berikut menata DAG arsip cadangan ke dalam dua grup. Grup muat berisi task baca_sumber dan simpan_arsip, sedangkan grup validasi berisi task periksa_isi dan catat_hasil. Fungsi muat mengembalikan XComArg dari task terakhirnya, dan objek itu diterima fungsi validasi sebagai argumen, sehingga hasil pemuatan mengalir ke tahap validasi tanpa deklarasi dependensi tambahan.

"""DAG arsip cadangan yang tertata dalam dua task group."""
import pendulum
from airflow.sdk import dag, task, task_group


@dag(
    dag_id="pemuatan_berkas",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def pemuatan_berkas():
    @task_group
    def muat():
        """Grup pengambilan dan penyimpanan berkas sumber."""

        @task
        def baca_sumber() -> list[str]:
            return ["data-2026-10-01.csv", "data-2026-10-02.csv"]

        @task
        def simpan_arsip(berkas: list[str]) -> str:
            return f"arsip dibuat dari {len(berkas)} berkas"

        return simpan_arsip(baca_sumber())

    @task_group
    def validasi(arsip: str):
        """Grup pemeriksaan hasil arsip."""

        @task
        def periksa_isi(arsip: str) -> bool:
            return arsip.startswith("arsip")

        @task
        def catat_hasil(boleh: bool) -> None:
            print(f"status validasi: {boleh}")

        catat_hasil(periksa_isi(arsip))

    validasi(muat())


dag = pemuatan_berkas()

Rantai validasi(muat()) bekerja dengan mekanisme yang sama seperti pemanggilan task bertingkat pada TaskFlow, yaitu nilai balik sebuah pemanggilan menjadi argumen pemanggilan berikutnya. Perbedaannya terletak pada objek yang dipindahkan, karena objek itu kini melintasi batas grup. Karena itu, Airflow menggambar garis dependensi dari task terakhir grup sumber ke task pertama grup tujuan pada graf.

Struktur DAG hasil perancangan diperiksa dengan mengimpor modulnya. Perhatikan bahwa daftar task kini memuat prefix nama grup.

# memeriksa struktur DAG beserta prefix task group
python3 -c "import pemuatan_berkas; d = pemuatan_berkas.dag; print(d.dag_id); print(sorted(t.task_id for t in d.tasks))"

pemuatan_berkas
['muat.baca_sumber', 'muat.simpan_arsip', 'validasi.catat_hasil', 'validasi.periksa_isi']

Prefix group_id pada nama task#

Secara bawaan, task_id setiap task anggota memperoleh prefix berupa group_id grupnya, dengan tanda titik sebagai pemisah. Prefix ini menjamin keunikan task_id di seluruh DAG sekaligus menyampaikan posisi task dalam hierarki, misalnya muat.baca_sumber jelas berada pada tahap pemuatan. Grup yang bersarang di dalam grup lain memperoleh prefix berlapis, yaitu nama grup induk diikuti nama grup anak.

Prefix dapat dimatikan dengan argumen prefix_group_id=False, baik pada decorator @task_group maupun pada kelas TaskGroup. Setelah prefix hilang, keunikan setiap task_id dan group_id menjadi tanggung jawab penulis DAG sepenuhnya. Contoh berikut membuat task di dalam grup muat dengan prefix yang dimatikan, lalu mencetak nama task yang dihasilkan.

from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import DAG, TaskGroup
import pendulum

with DAG(
    dag_id="uji_taskgroup_tanpa_prefix",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
) as dag:
    muat = TaskGroup("muat", prefix_group_id=False)
    t = EmptyOperator(task_id="proses", task_group=muat)
    print(t.task_id)

proses

Task anggota grup muat bernama proses tanpa nama grup di depannya. Apabila argumen prefix_group_id=False dihilangkan, perintah yang sama mencetak muat.proses.

Diagram berikut memperlihatkan bentuk DAG contoh pada UI. Dua grup digambar pada graf DAG, masing-masing menaungi dua task, dengan satu panah yang melintasi batas grup sebagai tanda dependensi lintas tahap.

Alternatif kelas TaskGroup#

Selain decorator, grup dapat dibuat dengan kelas TaskGroup yang diimpor dari airflow.sdk. Kelas ini menerima group_id sebagai argumen pertama, dan task yang dibuat di dalam context grup otomatis menjadi anggotanya. Cara ini digunakan apabila struktur grup ditentukan melalui kode, misalnya grup yang dibentuk di dalam perulangan dari daftar konfigurasi. Untuk grup statis, decorator @task_group lebih ringkas karena fungsi grup sekaligus menampung definisi seluruh task anggotanya.

from airflow.sdk import DAG, TaskGroup
import pendulum

with DAG(
    dag_id="uji_taskgroup",
    schedule="@once",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
) as dag:
    muat = TaskGroup("muat")
    print(muat.group_id)
    print(type(muat).__name__)

muat
TaskGroup

Bentuk visual pada Graph view#

Graph view menggambarkan struktur logis DAG, yaitu urutan task beserta dependensinya, dengan setiap simpul mewakili satu task dan setiap garis mewakili satu dependensi. Task group menambahkan satu lapisan pada gambar itu, yaitu kotak dengan batas putus-putus yang menaungi task anggotanya dan diberi judul nama grup. Panah dependensi antar task tetap digambar apa adanya, termasuk panah yang melintasi batas kotak apabila dua task dari grup berbeda saling bergantung.

Grup tidak menambah task maupun dag run baru. DAG pada topik ini tetap memiliki empat task sebagaimana keluaran inspeksi sebelumnya, sehingga pengaturan eksekusi seperti slot, retry, dan kebijakan pool tidak berubah karena keberadaan grup. Manfaatnya murni pada keterbacaan, yaitu graf yang memuat banyak tahapan tetap dapat dipindai karena setiap tahapan berbentuk satu kotak bernama. Pilih batas grup menurut batas tanggung jawab, misalnya tahap pengambilan data dipisahkan dari tahap pemeriksaannya, agar prefix pada task_id turut menyampaikan alur pipeline.

Kesalahan umum#

Mematikan prefix lalu menggunakan nama task yang sama. Tanpa prefix, dua grup yang sama-sama memuat task_id="proses" menghasilkan task_id ganda dan DAG gagal diparse. Beri nama unik pada setiap task, atau pertahankan prefix bawaan.

Tidak memanggil fungsi grup. Fungsi berdecorator @task_group yang hanya didefinisikan tetapi tidak pernah dipanggil tidak mendaftarkan grup apa pun pada DAG, sehingga task di dalamnya ikut hilang. Panggil fungsi grup di dalam generator DAG, sebagaimana validasi(muat()) pada contoh.

Menganggap grup sebagai unit eksekusi terpisah. Grup bukan DAG kedua dan tidak memiliki jadwal sendiri, sehingga pengaturan schedule, catchup, dan retries tetap berada di tingkat DAG atau task. Grup hanya menyusun tampilan dan hierarki nama.

Ringkasan#

• Decorator @task_group mengelompokkan task menjadi satu kotak pada Graph view tanpa mengubah perilaku eksekusi DAG.
• task_id anggota grup otomatis berprefix group_id, misalnya muat.baca_sumber, dan prefix dimatikan dengan prefix_group_id=False.
• Kelas TaskGroup dari airflow.sdk menyediakan cara yang sama untuk grup yang dibentuk melalui kode.
• Docstring fungsi grup menjadi tooltip grup pada UI.

Referensi#

• Dags — Apache Airflow Documentation — Bagian TaskGroups yang membahas decorator @task_group, prefix group_id, dan tooltip.
• Apache Airflow UI — Apache Airflow Documentation — Bagian Graph View yang menjelaskan penggambaran struktur logis DAG dan dependensinya.