KAL
Menu

Menulis DAG

Menulis DAG pertama dengan PythonOperator dan TaskFlow API.

Daftar isi

Topik ini membahas cara menulis DAG pertama pada Airflow versi 3 dengan dua gaya, yaitu gaya klasik berbasis PythonOperator dan gaya dekorator berbasis TaskFlow API, yakni gaya penulisan yang memakai dekorator dan pemanggilan fungsi biasa. Pembahasan diawali penempatan berkas DAG, dilanjutkan contoh lengkap untuk kedua gaya, lalu diakhiri cara menguji logika task terpisah dari Airflow.

Letak berkas dan penemuan DAG#

Airflow menemukan DAG dengan memindai folder yang dikonfigurasi melalui opsi dags_folder pada berkas airflow.cfg, sehingga menulis DAG berarti menempatkan satu berkas Python ke folder tersebut. Setiap berkas memuat satu atau lebih DAG yang dinyatakan lewat DAG(...) atau dekorator @dag, dan setiap DAG diberi dag_id yang unik. Argumen schedule menjadi rumus penjadwalannya, misalnya @daily, sedangkan start_date menentukan batas awal tanggal yang dijadwalkan. Contoh lama di internet kerap memakai argumen schedule_interval, dan pada versi 3 argumen itu sudah digantikan sepenuhnya oleh schedule.

Satu berkas boleh memuat lebih dari satu DAG, tetapi kebiasaan yang sehat adalah satu DAG per berkas dengan nama berkas yang mengikuti dag_id, sebagaimana dag_penjualan.py untuk rekap_penjualan. Berkas yang gagal diimpor tidak menghentikan Airflow, melainkan dilaporkan sebagai import error, yakni pesan kesalahan yang muncul saat berkas DAG dibaca. Daftarnya dapat ditelusuri lewat perintah airflow dags list-import-errors, dan kebiasaan memeriksa daftar tersebut setelah menyimpan berkas baru mempercepat penemuan kesalahan sintaks.

Sesaat setelah berkas tersimpan, dag processor memindainya dan DAG yang valid tampil di UI tanpa perlu menyalakan ulang apa pun. Untuk menguji satu DAG dari berkas tertentu, Airflow menyediakan perintah berikut yang menjalankan seluruh task secara lokal pada tanggal yang diminta.

bash 
# tidak-dijalankan
# menguji DAG dari berkas tertentu tanpa menyalakan scheduler
airflow dags test rekap_penjualan -f dags/dag_penjualan.py

Versi PythonOperator#

PythonOperator menjalankan fungsi Python biasa, yang diberikan melalui argumen python_callable. Contoh berikut membaca berkas sumber pada task pertama, lalu menuliskan total penjualan pada task kedua, dengan dependensi dinyatakan lewat operator >>.

python dag_penjualan.py 
# tidak-dijalankan
import datetime

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG

from transformasi import total_penjualan

FOLDER = "/opt/airflow/data/penjualan"


def baca_data() -> list[str]:
    """Membaca berkas sumber dan mengembalikan baris yang tidak kosong."""
    with open(f"{FOLDER}/sumber.csv") as berkas:
        return [baris for baris in berkas.read().splitlines() if baris]


def tulis_total(**konteks) -> None:
    """Menjumlahkan baris valid lalu menulis hasilnya ke berkas rekap."""
    data = konteks["ti"].xcom_pull(task_ids="baca_data")
    total = total_penjualan(data)
    with open(f"{FOLDER}/rekap.txt", "w") as berkas:
        berkas.write(f"total penjualan: {total}\n")


with DAG(
    dag_id="rekap_penjualan",
    schedule="@daily",
    start_date=datetime.datetime(2026, 1, 1),
    catchup=False,
) as dag:
    baca = PythonOperator(task_id="baca_data", python_callable=baca_data)
    tulis = PythonOperator(task_id="tulis_total", python_callable=tulis_total)
    baca >> tulis

Tiga hal pada contoh tersebut lazim ditemui pada DAG klasik. Impor PythonOperator berasal dari airflow.providers.standard.operators.python, karena pada versi 3 operator bawaan dipindahkan ke standard provider. Fungsi tulis_total menerima **konteks, yakni kamus informasi runtime yang disuntikkan Airflow, sehingga dapat mengambil hasil task pertama lewat xcom_pull. Nilai kembalian setiap task disimpan sebagai XCom, yakni mekanisme penukaran nilai kecil antar task, dan baris baca >> tulis menyatakan bahwa task tulis hanya boleh berjalan setelah task baca selesai.

Untuk nilai yang sudah diketahui saat DAG ditulis, masukan fungsi cukup dikirim melalui argumen op_kwargs tanpa perlu membaca XCom, sebagaimana contoh resmi yang mengirim nilai tetap ke setiap task. XCom hanya dibutuhkan ketika nilainya baru tersedia saat eksekusi, misalnya banyak baris yang dihitung task sebelumnya.

Versi TaskFlow#

TaskFlow API menyusun DAG dari fungsi yang diberi dekorator, sehingga dependensi dan pengaliran nilai menjadi bagian dari pemanggilan fungsi. Berkas berikut mengerjakan rekap yang sama dengan gaya tersebut.

python dag_taskflow.py 
# tidak-dijalankan
import datetime
from pathlib import Path

from airflow.sdk import dag, task

from transformasi import total_penjualan

FOLDER = Path("/opt/airflow/data/penjualan")


@dag(
    dag_id="rekap_penjualan_taskflow",
    schedule="@daily",
    start_date=datetime.datetime(2026, 1, 1),
    catchup=False,
)
def rekap_penjualan():
    @task
    def baca_data() -> list[str]:
        """Membaca berkas sumber dan mengembalikan baris yang tidak kosong."""
        berkas = FOLDER / "sumber.csv"
        return [baris for baris in berkas.read_text().splitlines() if baris]

    @task
    def tulis_total(data: list[str]) -> int:
        """Menjumlahkan baris valid lalu menulis hasilnya ke berkas rekap."""
        total = total_penjualan(data)
        (FOLDER / "rekap.txt").write_text(f"total penjualan: {total}\n")
        return total

    tulis_total(baca_data())


rekap_penjualan()

Baris tulis_total(baca_data()) memuat dua perintah sekaligus, yaitu mendaftarkan kedua task dan menyatakan bahwa keluaran baca_data menjadi masukan tulis_total. Nilai kembaliannya dipindahkan melalui XCom secara otomatis, sehingga kode xcom_pull pada gaya klasik tidak diperlukan lagi. Nama fungsi menjadi task_id, dan pemanggilan fungsi yang diberi dekorator @task tidak mengeksekusi apa pun, karena pemanggilan itu hanya mendaftarkan task ke DAG pemiliknya.

Kedua gaya menghasilkan DAG yang setara, sehingga pilihannya adalah soal keterbacaan dan kesesuaian dengan kode yang sudah ada. TaskFlow lebih ringkas untuk pengaliran nilai antar task dan hidup berdampingan dengan penanda tipe Python, sedangkan gaya klasik tetap perlu dikuasai karena banyak contoh di dokumentasi proyek lain memakainya. Mampu membaca keduanya memudahkan penilaian DAG warisan tanpa harus mengganti seluruh isinya terlebih dahulu.

Menguji logika tanpa Airflow#

Kedua DAG di atas memanggil total_penjualan dari modul transformasi.py, yang tidak mengimpor apa pun dari Airflow. Pemisahan semacam ini membuat logika transformasi dapat dijalankan dan diperiksa dengan Python biasa, sebagaimana tereksekusi berikut.

python transformasi.py 
"""Logika transformasi yang dipakai task pada DAG rekap penjualan.

Modul ini sengaja tidak mengimpor apa pun dari Airflow sehingga dapat diuji
dengan Python biasa, baik secara manual maupun lewat pengujian otomatis.
"""


def baris_valid(baris: str) -> bool:
    """Baris valid bila berbentuk tanggal,jumlah dan jumlahnya berupa angka."""
    bagian = baris.strip().split(",")
    return len(bagian) == 2 and len(bagian[0]) == 10 and bagian[1].isdigit()


def total_penjualan(data: list[str]) -> int:
    """Menjumlahkan kolom jumlah dari baris-baris yang valid."""
    angka = []
    for baris in data:
        if not baris_valid(baris):
            continue
        angka.append(int(baris.strip().split(",")[1]))
    return sum(angka)


def hitung_terabaikan(data: list[str]) -> int:
    """Menghitung banyak baris yang tidak valid dan dikesampingkan."""
    return sum(1 for baris in data if not baris_valid(baris))


if __name__ == "__main__":
    contoh = [
        "2026-03-02,120",
        "2026-03-02,tiga",
        "2026-03-03,87",
        "",
        "2026-03-03,41",
    ]
    print(f"baris valid         : {len(contoh) - hitung_terabaikan(contoh)}")
    print(f"baris dikesampingkan: {hitung_terabaikan(contoh)}")
    print(f"total penjualan     : {total_penjualan(contoh)}")
text Keluaran 
baris valid         : 3
baris dikesampingkan: 2
total penjualan     : 248

Keluaran mengonfirmasi bahwa dua baris rusak dikesampingkan dan jumlah baris validnya dihitung benar. Ketika fungsi ini berubah, pemeriksaan cukup dilakukan pada modulnya sebelum DAG dipasang, karena perilaku task sepenuhnya ditentukan fungsi murni yang dipanggilnya. Modul semacam ini juga siap diperiksa oleh alat pengujian otomatis, karena fungsi-fungsinya menerima daftar string dan mengembalikan angka tanpa menyentuh jaringan maupun berkas konfigurasi, sehingga perubahan yang salah tertangkap sebelum task berjalan di penjadwal. Kebiasaan kecil ini memangkas waktu pencarian penyebab pada saat pipeline sudah dipasang, dan berkas DAG tetap ramping karena hanya berisi pendaftaran task beserta dependensinya.

Kesalahan umum#

  1. Mengimpor operator dengan jalur versi 2.x. Penyebabnya, contoh lama menuliskan from airflow.operators.python import PythonOperator, sedangkan versi 3 memindahkan operator ke standard provider. Perbaikannya, impor dari airflow.providers.standard.operators.python dan ambil dekorator TaskFlow dari airflow.sdk.
  2. Memakai argumen schedule_interval pada DAG baru. Penyebabnya, contoh lama dari versi 2.x masih banyak ditemukan, sedangkan argumen itu sudah digantikan oleh schedule. Perbaikannya, tulis schedule="@daily" atau pola cron yang dibutuhkan.
  3. Memanggil satu fungsi @task dua kali dalam DAG yang sama. Penyebabnya, setiap pemanggilan membuat task baru dengan task_id yang diambil dari nama fungsinya, sehingga keduanya bertabrakan. Perbaikannya, panggil sekali, simpan hasilnya ke satu variabel, lalu gunakan ulang variabel tersebut.
  4. Menanamkan logika transformasi di dalam fungsi task. Penyebabnya, logika yang bercampur dengan panggilan Airflow sulit diuji dan sulit digunakan ulang. Perbaikannya, pindahkan logika ke modul murni seperti transformasi.py dan uji modulnya secara terpisah.

Ringkasan#

  • DAG ditulis sebagai berkas Python pada folder dags_folder, dengan schedule dan start_date sebagai argumen penjadwalan utamanya.
  • Gaya klasik memakai PythonOperator dengan python_callable, **konteks, dan xcom_pull, sedangkan TaskFlow memakai @dag dan @task dengan dependensi yang tersusun dari pemanggilan fungsi.
  • Logika transformasi sebaiknya dipisahkan ke modul murni agar dapat diuji dengan Python biasa tanpa memasang Airflow.

Referensi#