KAL
Menu

DAG pertama

Menulis DAG pertama dengan TaskFlow API, menjalankannya, dan membaca status task pada UI.

Daftar isi

DAG pertama ditulis dengan TaskFlow API, yaitu gaya penulisan DAG yang menggunakan decorator Python untuk mendefinisikan task. Decorator adalah penanda berbentuk @ yang membungkus fungsi dan menambahkan perilaku padanya. Topik ini membahas penulisan berkas DAG, pemeriksaan struktur DAG melalui Python dan CLI, pendaftaran DAG ke basis data metadata, serta pembacaan status task pada CLI dan antarmuka web. DAG pertama yang dihasilkan menyimpan hasil perhitungan sederhana, dan setiap tahap eksekusinya dapat dijelaskan baris demi baris.

TaskFlow API adalah antarmuka penulisan DAG yang menggunakan decorator Python. Fungsi yang dibungkus decorator @dag menjadi definisi DAG, sedangkan fungsi yang dibungkus decorator @task menjadi satu task. Objek DAG baru tercipta ketika fungsi yang dibungkus @dag dipanggil di akhir berkas. Pola ini menjadi gaya baku pada Airflow 3 dan tetap digunakan berdampingan dengan gaya klasik berbasis operator.

Berkas dag_halo.py berikut adalah DAG pertama yang diperiksa dan dijalankan sepanjang topik ini. Tiga task di dalamnya membentuk satu rantai sederhana, yaitu menyiapkan deret nilai, menghitung statistiknya, lalu menampilkan hasil perhitungan ke log task.

python dag_halo.py 
from airflow.sdk import dag, task


@dag(
    dag_id="dag_halo",
    schedule=None,
    description="DAG latihan: menyiapkan deret nilai lalu menghitung statistik sederhana.",
)
def dag_halo():
    @task
    def siapkan_angka() -> list[int]:
        """Menyiapkan deret nilai yang akan dihitung."""
        return [4, 8, 15, 16, 23, 42]

    @task
    def hitung_statistik(angka: list[int]) -> dict:
        """Menghitung jumlah, rata-rata, dan nilai terbesar."""
        return {
            "jumlah": sum(angka),
            "rata_rata": sum(angka) / len(angka),
            "maksimum": max(angka),
        }

    @task
    def tampilkan(statistik: dict) -> None:
        """Menampilkan hasil perhitungan ke log task."""
        print(f"Statistik: {statistik}")

    tampilkan(hitung_statistik(siapkan_angka()))


dag_halo()

Argumen schedule bernilai None, sehingga DAG tidak memiliki jadwal dan hanya berjalan ketika dipicu secara manual. Nilai kembalian setiap task diteruskan secara otomatis menjadi argumen bagi task berikutnya, tanpa kode penghubung tambahan. Karena itu, baris terakhir cukup memanggil task terakhir dengan hasil task sebelumnya. Nama task diambil dari nama fungsinya, yaitu siapkan_angka, hitung_statistik, dan tampilkan. Fungsi tampilkan memanggil print sehingga hasil perhitungan muncul pada log saat task benar-benar dieksekusi.

Menyiapkan lingkungan latihan#

Perintah CLI Airflow membaca lokasi kerja dari variabel lingkungan AIRFLOW_HOME. Nilai variabel tersebut berupa folder yang memuat konfigurasi, basis data metadata, dan folder dags. Latihan pada topik ini menggunakan folder terpisah di /tmp/ra-dag-pertama agar tidak bercampur dengan instalasi lain. Pemuatan contoh DAG bawaan juga dinonaktifkan supaya daftar DAG pada CLI benar-benar kosong sebelum DAG latihan terdaftar.

bash 
# lingkungan latihan terpisah agar tidak bercampur dengan instalasi lain
export AIRFLOW_HOME=/tmp/ra-dag-pertama
# contoh DAG bawaan dimatikan supaya daftar DAG benar-benar kosong
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# uji bahwa berkas DAG dapat diparse tanpa keluaran apa pun
python3 dag_halo.py

Perintah terakhir memuat berkas DAG dan mendefinisikan seluruh objek di dalamnya. Apabila terdapat kesalahan sintaks atau impor, pesan error akan tampil, sedangkan berkas yang valid selesai dieksekusi tanpa keluaran apa pun. Keadaan senyap seperti ini menjadi indikator cepat bahwa struktur DAG dapat dibaca Airflow. Pada penggunaan sehari-hari, pemrosesan berkas DAG dilakukan oleh komponen DAG processor secara berkala. Latihan di sini memicu pemrosesan tersebut melalui CLI agar setiap langkah dapat diamati.

bash 
# siapkan folder dags lalu salin berkas DAG ke dalamnya
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_halo.py "$AIRFLOW_HOME/dags/"

Memeriksa struktur DAG#

Struktur DAG yang telah diparse dapat diperiksa tanpa menjalankan task sama sekali. Skrip satu baris berikut memuat modul DAG, memanggil fungsi DAG untuk memperoleh objeknya, lalu mencetak dag_id dan seluruh task_id dalam urutan alfabetis.

bash 
# muat modul DAG lalu cetak dag_id dan seluruh task_id
python3 -c "import dag_halo; d = dag_halo.dag_halo(); print(d.dag_id); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
dag_halo
['hitung_statistik', 'siapkan_angka', 'tampilkan']

Keluaran memuat dua baris sesuai harapan. Baris pertama berisi dag_id yang sesuai dengan argumen pada decorator, sedangkan baris kedua menampilkan tiga task sesuai nama fungsinya masing-masing. Pemeriksaan semacam ini juga menjadi dasar pengujian mutu DAG secara otomatis karena cepat dan tidak memerlukan eksekusi task.

Mendaftarkan DAG ke basis data metadata#

CLI membaca daftar DAG dari basis data metadata, yaitu basis data internal Airflow yang menyimpan DAG dan status eksekusinya, bukan langsung dari berkas. Lokasi folder dags diambil dari konfigurasi core.dags_folder, dan nilainya diturunkan dari AIRFLOW_HOME yang telah ditetapkan.

bash 
# tunjukkan folder dags yang dibaca Airflow
airflow config get-value core dags_folder
text Keluaran 
/tmp/ra-dag-pertama/dags

Dua langkah berikut menyiapkan penyimpanan DAG pada basis data metadata. Perintah db migrate membuat skema basis data metadata, sedangkan dags reserialize memproses berkas di folder dags menjadi DAG berserialisasi, yaitu representasi struktur DAG yang tersimpan pada basis data. Pada instalasi nyata, kedua pekerjaan ini dilakukan oleh scheduler dan DAG processor secara terus-menerus, sehingga DAG baru otomatis terdaftar beberapa saat setelah berkasnya disalin.

bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
airflow db migrate > /dev/null 2>&1
bash 
# proses berkas DAG menjadi DAG berserialisasi di basis data
airflow dags reserialize > /dev/null 2>&1

Setelah terdaftar, struktur DAG dapat ditanyakan kembali melalui CLI. Perintah tasks list menampilkan task milik satu DAG sesuai urutan alfabetis, dan hasilnya cocok dengan pemeriksaan Python pada bagian sebelumnya.

bash 
# daftar task milik dag_halo
airflow tasks list dag_halo
text Keluaran 
hitung_statistik
siapkan_angka
tampilkan
bash 
# daftar DAG yang terdaftar di basis data metadata
airflow dags list
text Keluaran 
dag_id   | fileloc                              | owners  | is_paused | bundle_name | bundle_version
=========+======================================+=========+===========+=============+===============
dag_halo | /tmp/ra-dag-pertama/dags/dag_halo.py | airflow | True      | dags-folder | None

Tabel keluaran memuat beberapa kolom yang penting dipahami. Kolom is_paused bernilai True karena DAG yang baru terdaftar dijeda secara default, sehingga scheduler tidak langsung menjalankannya sampai diaktifkan lewat UI atau perintah dags unpause. Kolom bundle_name menyatakan sumber berkas DAG, yaitu bundle bawaan bernama dags-folder yang isinya diambil dari folder dags.

Menjalankan DAG dan membaca status task#

Perintah dags test menjalankan satu dag run, yaitu satu kali eksekusi DAG beserta seluruh task di dalamnya. Eksekusi berjalan secara lokal tanpa memerlukan scheduler maupun worker terpisah. Perintah ini cocok untuk menguji DAG pertama sebelum dipasang pada lingkungan penggunaan nyata.

bash 
# jalankan seluruh task DAG sekali secara lokal
# log panjang dan bertanda waktu, sehingga diarahkan ke berkas
airflow dags test dag_halo 2026-11-04 > "$AIRFLOW_HOME/dags-test.log" 2>&1

Keluaran perintah ini panjang dan bertanda waktu, sehingga diarahkan ke berkas. Pada bagian akhir log, ketiga task berurutan berstatus success dan dag run ditutup dengan status yang sama. Argumen tanggal memberi penanda waktu pada dag run yang dihasilkan. Baris keluaran print milik task tampilkan diambil dari berkas log dan memuat Statistik: {'jumlah': 108, 'rata_rata': 18.0, 'maksimum': 42}, sehingga isi perhitungan dapat diverifikasi tanpa membuka UI.

bash 
# ambil baris keluaran task dari berkas log
export AIRFLOW_HOME=/tmp/ra-dag-pertama
grep -o "Statistik: {'jumlah': 108, 'rata_rata': 18.0, 'maksimum': 42}" "$AIRFLOW_HOME/dags-test.log"
text Keluaran 
Statistik: {'jumlah': 108, 'rata_rata': 18.0, 'maksimum': 42}

Status yang sama dapat dibaca pada antarmuka web. Halaman DAG menyediakan Grid view untuk melihat status task lintas dag run dalam bentuk kisi, serta Graph view untuk melihat graf dependensi antar task. Setiap sel menyatakan satu task instance, yaitu salinan task pada satu dag run, dengan warna sesuai statusnya, misalnya hijau untuk success dan merah untuk failed. Mengeklik sel membuka log task, sehingga keluaran task tampilkan dapat dibaca langsung dari UI.

Ringkasan#

  • DAG TaskFlow didefinisikan dengan @dag dan @task, lalu diwujudkan dengan memanggil fungsi DAG di akhir berkas.
  • Berkas DAG harus berada di folder dags dan diproses menjadi DAG berserialisasi sebelum terlihat oleh CLI maupun scheduler.
  • airflow dags test menjalankan satu dag run secara lokal, sedangkan status task dibaca pada Grid view melalui warna sel atau melalui log task.

Referensi#