KAL
Menu

Mutu DAG

Menjaga mutu DAG dengan linting, unit test, dan test tingkat DAG.

Daftar isi

DAG yang gagal diparse membuat pipeline tidak muncul sama sekali. DAG yang berhasil diparse tetapi logikanya salah menghasilkan data yang salah tanpa pesan error apa pun. Topik ini membahas tiga lapis pemeriksaan mutu yang menutup kedua risiko tersebut. Ketiganya adalah pemeriksaan impor pada waktu berkas DAG dimuat, unit test atas fungsi murni yang dipisah dari DAG, dan test tingkat DAG yang menegaskan strukturnya. Seluruh pemeriksaan dieksekusi nyata pada topik ini beserta keluaran aslinya.

Pemeriksaan impor pada waktu muat#

Airflow memuat setiap berkas di folder dags sebagai modul Python, dan berkas yang gagal dimuat tidak menghasilkan DAG apa pun. Kegagalan ini mudah luput dari perhatian karena scheduler melanjutkan pekerjaannya dengan DAG lain, sementara DAG yang bermasalah hanya meninggalkan catatan error pada basis data metadata. Perintah airflow dags reserialize memindai folder dags dan mencatat setiap kegagalan muat, lalu airflow dags list-import-errors menampilkannya.

Berkas berikut sengaja dibuat rusak, yaitu mengimpor modul yang tidak ada, sebagaimana lazim terjadi setelah nama modul diubah tanpa memperbarui seluruh berkas DAG.

python dag_rusak.py 
# tidak-dijalankan
"""DAG rusak sengaja untuk memperlihatkan import errors.

Baris impor modul yang tidak ada membuat berkas ini gagal dimuat,
dan pesan kesalahannya muncul pada `airflow dags list-import-errors`.
"""
import pendulum
import modul_tidak_ada
from airflow.sdk import dag, task


@dag(dag_id="dag_rusak", schedule=None, description="Sengaja gagal diimpor.")
def dag_rusak():
    @task
    def kosong() -> None:
        """Task yang tidak pernah berjalan."""
        print("tidak pernah berjalan")

    kosong()


dag_rusak()

Pemindaian dijalankan pada lingkungan latihan yang berisi DAG rusak tersebut berdampingan dengan DAG sehat bernama rekap_absensi. Perintah pemeriksaan kemudian menunjukkan bahwa DAG sehat tetap dimuat, sementara DAG rusak tercatat sebagai error.

python dag_rekap_absensi.py 
"""DAG rekap absensi sekolah sebagai objek pemeriksaan mutu.

Memuat satu fungsi murni yang diuji unit test dan satu DAG dua task
yang diuji pada tingkat DAG.
"""
import pendulum
from airflow.sdk import dag, task


def hitung_persen_hadir(hadir: int, total: int) -> int:
    """Menghitung persen kehadiran; menolak total yang tidak sah."""
    if total <= 0:
        raise ValueError("total siswa harus lebih dari nol")
    return round(100 * hadir / total)


@dag(
    dag_id="rekap_absensi",
    schedule=None,
    start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
    default_args={"retries": 2},
    description="Rekap persen kehadiran per kelas.",
)
def rekap_absensi():
    @task
    def baca_absensi() -> dict:
        """Membaca pasangan hadir dan total siswa per kelas."""
        return {"A": (28, 30), "B": (26, 29)}

    @task
    def tulis_rekap(absensi: dict) -> None:
        """Menuliskan rekap persen kehadiran setiap kelas."""
        for kelas, (hadir, total) in absensi.items():
            print(f"kelas {kelas}: {hitung_persen_hadir(hadir, total)}%")

    tulis_rekap(baca_absensi())


rekap_absensi()
bash 
# lingkungan latihan terpisah untuk topik ini
export AIRFLOW_HOME=/tmp/ra-mutu-dag
export AIRFLOW__CORE__LOAD_EXAMPLES=False
bash 
# salin kedua berkas DAG ke folder yang dipindai Airflow
export AIRFLOW_HOME=/tmp/ra-mutu-dag
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_rekap_absensi.py dag_rusak.py "$AIRFLOW_HOME/dags/"
bash 
# siapkan skema basis data metadata lalu pindai folder dags
# log kedua perintah bertanda waktu, sehingga diarahkan ke berkas
export AIRFLOW_HOME=/tmp/ra-mutu-dag
airflow db migrate > /dev/null 2>&1
airflow dags reserialize > "$AIRFLOW_HOME/reserialize.log" 2>&1

Perintah list-import-errors menampilkan satu baris untuk setiap berkas yang gagal dimuat, lengkap dengan traceback singkatnya. Baris log internal disaring dengan grep -v agar tabelnya bersih.

bash 
# tampilkan daftar berkas DAG yang gagal dimuat
export AIRFLOW_HOME=/tmp/ra-mutu-dag
airflow dags list-import-errors 2>&1 | grep -v alembic | head -n 10
text Keluaran 
bundle_name | filepath     | error
============+==============+===============================================================================
dags-folder | dag_rusak.py | Traceback (most recent call last):
            |              |   File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
            |              |   File "/tmp/ra-mutu-dag/dags/dag_rusak.py", line 8, in <module>
            |              |     import modul_tidak_ada
            |              | ModuleNotFoundError: No module named 'modul_tidak_ada'
            |              |

Tabel menyebut dag_rusak.py beserta penyebab pastinya, yaitu ModuleNotFoundError pada baris 8. DAG rekap_absensi tidak muncul pada daftar ini karena dimuat tanpa error, dan pemeriksaan yang sama dijalankan ulang setiap kali berkas DAG di folder dags berubah. Dalam pipeline pengembangan, perintah ini dipasang sebagai gerbang pada CI sehingga berkas DAG yang gagal dimuat tidak pernah sampai ke lingkungan production.

Unit test fungsi murni#

Logika perhitungan yang ditulis di dalam task sulit diuji karena pengujian tersebut berarti menjalankan Airflow. Cara paling bersih adalah memisahkan logika ke dalam fungsi murni di modul DAG, sebagaimana fungsi hitung_persen_hadir pada dag_rekap_absensi.py, lalu mengujinya dengan test biasa. Fungsi murni menerima masukan dan mengembalikan keluaran tanpa menyentuh konteks Airflow, sehingga ujinya cepat dan tidak memerlukan basis data metadata.

Berkas test_mutu_fungsi.py menguji dua hal yang menentukan, yaitu hasil perhitungan yang benar dan penolakan total yang tidak sah. Kasus penolakan diuji dengan pytest.raises yang menegaskan exception tertentu benar-benar dimunculkan.

python test_mutu_fungsi.py 
# tidak-dijalankan
"""Unit test fungsi murni yang dipisah dari DAG.

Berkas ini dijalankan melalui pytest dari folder tempat modul DAG
berada, bukan langsung dengan python.
"""
import pytest

from dag_rekap_absensi import hitung_persen_hadir


def test_perhitungan_normal():
    assert hitung_persen_hadir(27, 30) == 90
    assert hitung_persen_hadir(1, 3) == 33


def test_total_nol_ditolak():
    with pytest.raises(ValueError):
        hitung_persen_hadir(27, 0)


def test_total_negatif_ditolak():
    with pytest.raises(ValueError):
        hitung_persen_hadir(27, -5)

Test tingkat DAG#

Pemeriksaan berikutnya menargetkan struktur DAG itu sendiri. Test tingkat DAG memuat modul DAG tanpa mengeksekusi task apa pun, lalu menegaskan sifat DAG yang disepakati tim, misalnya jumlah task, jadwal, dan kebijakan retry. Test semacam ini menangkap perubahan tak sengaja, misalnya task yang terhapus saat refactor atau default_args yang kehilangan retries. Kedua berkas test pada topik ini menggunakan awalan test_ sehingga pytest dapat menemukannya dari nama berkas, dan keduanya dijalankan dari folder tempat modul DAG berada agar pernyataan impor modul DAG berhasil. Susunan yang sama berlaku pada proyek DAG pada umumnya, yaitu test fungsi murni dan test tingkat DAG berdampingan dalam satu folder.

python test_dag_rekap.py 
# tidak-dijalankan
"""Test tingkat DAG untuk DAG rekap_absensi.

Berkas ini dijalankan melalui pytest dari folder tempat modul DAG
berada, bukan langsung dengan python.
"""
import dag_rekap_absensi


def test_struktur_dag():
    d = dag_rekap_absensi.rekap_absensi()
    assert d.dag_id == "rekap_absensi"
    assert d.schedule is None
    assert len(d.tasks) == 2


def test_setiap_task_mempunyai_retry():
    d = dag_rekap_absensi.rekap_absensi()
    for t in d.tasks:
        assert t.retries >= 1, f"task {t.task_id} tanpa retry"

Kedua berkas test dijalankan sekaligus dengan pytest -v. Keluaran disaring agar hanya baris status test yang tampil.

bash 
# jalankan seluruh test mutu DAG dari folder berkas DAG berada
python3 -m pytest test_mutu_fungsi.py test_dag_rekap.py -v 2>&1 | grep "PASSED"
text Keluaran 
test_mutu_fungsi.py::test_perhitungan_normal PASSED                      [ 20%]
test_mutu_fungsi.py::test_total_nol_ditolak PASSED                       [ 40%]
test_mutu_fungsi.py::test_total_negatif_ditolak PASSED                   [ 60%]
test_dag_rekap.py::test_struktur_dag PASSED                              [ 80%]
test_dag_rekap.py::test_setiap_task_mempunyai_retry PASSED               [100%]

Lima test lulus seluruhnya, tiga pada fungsi murni dan dua pada tingkat DAG. Ketika salah satu tegasan dilanggar, misalnya retries dihapus dari default_args, test test_setiap_task_mempunyai_retry berstatus failed dengan pesan yang menunjuk task yang bermasalah. Dengan demikian, pelanggaran ketentuan ditemukan sebelum DAG diterapkan ke production, bukan setelah run production berjalan.

Kesalahan umum#

  • Menempatkan logika berat di dalam fungsi task dan mengujinya melalui eksekusi DAG. Eksekusi DAG memakan waktu belasan detik per run dan keluarannya sulit ditelusuri, sedangkan fungsi murni diuji dalam hitungan detik untuk seluruh kasus batasnya. Pindahkan logika ke fungsi modul, lalu panggil fungsinya dari task.
  • Menulis test yang memuat seluruh folder dags sekaligus. Satu berkas DAG yang gagal dimuat membuat seluruh suite gagal sebelum test berjalan, dan penyebabnya tersamar. Muat DAG secara eksplisit dengan mengimpor modulnya, sebagaimana import dag_rekap_absensi pada contoh, sehingga kegagalan menunjuk berkas yang tepat.
  • Memperlakukan pemeriksaan impor sebagai pekerjaan manual. Pemeriksaan yang hanya mengandalkan ingatan pada akhirnya terlewat. Jadikan airflow dags list-import-errors sebagai langkah otomatis pada CI agar berkas rusak tertahan sebelum menjangkau folder dags production.

Ringkasan#

  • airflow dags reserialize mencatat kegagalan muat berkas DAG dan airflow dags list-import-errors menampilkannya beserta traceback singkatnya, sehingga berkas rusak ditemukan sebelum deployment.
  • Logika perhitungan dipisahkan ke fungsi murni di modul DAG agar dapat diuji dengan pytest tanpa mengeksekusi Airflow, termasuk kasus penolakan masukan tidak sah melalui pytest.raises.
  • Test tingkat DAG memuat modul DAG dan menegaskan strukturnya, misalnya jumlah task, jadwal, dan kebijakan retry, sehingga perubahan tak sengaja tertangkap pada waktu test.
  • Ketiga pemeriksaan dijalankan otomatis pada CI, dan keluaran contoh pada topik ini memperlihatkan lima test lulus beserta satu berkas DAG rusak yang tertangkap.

Referensi#