Bagian 8: Memantau Eksekusi

Pipeline yang berjalan sendiri di tengah malam hanya berguna bila pagi harinya kita bisa menjawab tiga pertanyaan dengan cepat: apakah semalam sukses, sudah sampai tanggal berapa, dan bila gagal di mana gagalnya. Airflow menjawabnya lewat UI dan log; proyek kita menambahkan satu lapisan milik sendiri — state watermark yang bisa dibaca tanpa membuka UI sama sekali. Bagian ini memetakan keduanya, dengan UI dijelaskan sebagai prosa berlabel jelas karena Airflow tidak terpasang, sementara state dibuktikan dengan eksekusi nyata.

Pemandangan di UI Airflow #

Pada deployment sungguhan, empat layar UI menjadi ruang kerja harian operator — deskripsi berikut adalah perilaku yang diharapkan, bukan tangkapan layar yang kita jalankan:

• Grid view: baris per DAG run dengan kolom per task; satu lirikan menunjukkan status success, failed, upstream_failed, atau skipped untuk setiap run dan task, termasuk run hasil catchup pada Bagian 7.
• Graph view: graf dependency persis seperti mermaid pada Bagian 4, dengan node yang mewarnai diri sesuai status task terakhir.
• Log task: setiap print pada task kita ([ekstrak] ..., [transform] ..., [muat] ...) menjadi baris log yang bisa dibaca per percobaan (attempt), berdampingan dengan log internal Airflow.
• Riwayat percobaan: untuk task yang di-retry pada Bagian 6, setiap attempt tampil sebagai entri terpisah sehingga pola “gagal dua kali lalu sukses” terlihat.

Membaca log task punya pola kerjanya sendiri. Operator biasanya membuka run yang merah, memilih task berstatus failed paling hulu di Graph view, lalu membaca log percobaan terakhir dari bawah ke atas — traceback Python ada di ujung berkas, dan baris berprefiks [ekstrak]/[muat] milik kita memberi konteks data pada momen sebelum galat. Untuk task yang upstream_failed, lognya nyaris kosong karena task tidak pernah berjalan; yang perlu dibaca adalah task asal kegagalan di hulunya.

Yang tidak ada di UI: isi zona data kita. UI tahu bahwa task muat sukses, tetapi tidak tahu berapa omzet hari itu — untuk itu kita punya lapisan berikutnya.

State sebagai Sumber Kebenaran #

Task laporan sengaja tidak membaca ulang zona raw maupun gold; task tersebut membaca catatan hasil eksekusi dari state:

def laporan_dari_state(state_path: str | Path, tanggal: str) -> str:
    """Bangun baris laporan untuk satu tanggal dari state saja.

    Dipakai task laporan: tidak menyentuh zona raw maupun gold, cukup membaca
    catatan hasil eksekusi yang tersimpan di state.
    """
    state = state_etl.baca_state(state_path)
    info = state["terolah"].get(tanggal)
    if info is None:
        return f"[{tanggal}] belum ada catatan di state"
    return laporan_run(_hasil_dari_info(info, tanggal, info["status"], state["watermark"]))

Keandalan lapisan ini bergantung pada satu properti: state tidak pernah terbaca dalam keadaan setengah tertulis. Penulisan dilakukan secara atomik:

def simpan_state(path: str | Path, state: dict) -> Path:
    """Simpan state secara atomik: tulis ke .tmp lalu ganti dengan os.replace."""
    berkas = Path(path)
    berkas.parent.mkdir(parents=True, exist_ok=True)
    sementara = berkas.with_name(berkas.name + ".tmp")
    with sementara.open("w", encoding="utf-8") as f:
        json.dump(state, f, indent=2, sort_keys=True)
        f.write("\n")
    os.replace(sementara, berkas)
    return berkas

State ditulis dulu ke berkas .tmp baru diganti ke lokasi asli lewat os.replace — operasi penggantian nama bersifat atomik pada sistem berkas, sehingga pembaca (task laporan lain, atau mata operator) tidak pernah melihat JSON terpotong walau penulisannya terganggu.

Skenario pemantauan berikut dijalankan nyata oleh skrip demonstrasi di luar bundel proyek: tiga partisi tersedia, tetapi hanya 02 November yang sempat diproses (misalnya scheduler mati satu hari), lalu operator bertanya “sudah sampai mana?”:

Output yang Dihasilkan:

== pertanyaan operator: sudah sampai mana pipeline ini? ==
watermark      : 2026-11-02
tanggal terolah: ['2026-11-02']
tanggal tertunda (jendela 02..04): ['2026-11-03', '2026-11-04']

== laporan per tanggal cukup dari state, tanpa menyentuh data ==
[2026-11-02] sukses: 7 baris mentah -> 4 valid, 2 ditolak, 1 duplikat; 3 kota, omzet Rp843,000 -> summary_harian_2026-11-02.csv
[2026-11-03] belum ada catatan di state
[2026-11-04] belum ada catatan di state

== berkas gold yang sudah dihasilkan ==
summary_harian_2026-11-02.csv

Tiga baris pertama menjawab pertanyaan tingkat kesehatan: watermark berhenti di 02 November dan dua tanggal tertunda — informasi yang langsung mengarahkan ke jalankan_backfill pada Bagian 7. Laporan per tanggal memberi detail angka tanpa membuka berkas data, dan daftar berkas gold membenarkan bahwa hanya satu partisi yang benar-benar ditulis. Perhatikan bahwa fungsi laporan_dari_state mengembalikan kalimat eksplisit untuk tanggal tanpa catatan, bukan galat — operator memindai daftar, bukan menangkap exception.

alarm_kegagalan dan Callback #

Saluran berikutnya adalah peringatan aktif. Task alarm_kegagalan pada berkas DAG berjalan begitu salah satu task hulu gagal:

    @task(trigger_rule="one_failed")
    def alarm_kegagalan(ds=None) -> None:
        """Stub alert: ganti print dengan notifikasi nyata di produksi."""
        print(f"[alarm] {DAG_ID} gagal memproses partisi {ds} - periksa log task")

Isi task masih berupa print (stub) agar bisa dibaca di log; di produksi, isi inilah tempat memanggil webhook Slack, mengirim email, atau menulis insiden. Alternatif yang tak kalah umum adalah on_failure_callback pada default_args — sebuah fungsi yang dipanggil Airflow saat task gagal tanpa perlu menjadi task tersendiri. Keduanya bekerja pada momen yang sama: setelah seluruh percobaan retry pada Bagian 6 habis.

Pemilihan antara keduanya soal selera dan kebutuhan: task alarm_kegagalan tampil di graf dan riwayat eksekusinya tercatat sebagai run, sedangkan callback lebih ringan tetapi hanya meninggalkan jejak di log.

Peta Observability #

Kolom terakhir menegaskan batas yang jujur: state kita menjawab pertanyaan tentang data dan progres, sedangkan pertanyaan tentang mesin (durasi, percobaan, antrian executor) hanya bisa dijawab Airflow itu sendiri.

Ritual Pagi Operator #

Kedua sumber tadi menyusun urutan kerja yang hemat waktu. Pagi hari, setelah run malam seharusnya selesai:

1. Baca state lebih dulu — watermark dan daftar tanggal tertunda menjawab “sehat atau tidak” dalam satu lirikan; ini bisa dari skrip, tanpa membuka browser.
2. Buka UI hanya bila ada yang merah — Grid view menunjukkan task mana yang gagal dan pada attempt ke berapa.
3. Baca log task yang gagal dari percobaan terakhir; baris berprefiks [ekstrak]/[transform]/[muat] memberi konteks data sebelum traceback.
4. Putuskan tindak lanjut — partisi telat berarti tunggu retry berikutnya; berkas rusak berarti perbaiki sumber lalu jalankan ulang tanggal itu; lalu catat insingennya bila berulang.

Urutan ini sengaja dimulai dari yang termurah (membaca satu berkas JSON) ke yang termahal (menyelidiki log percobaan). Alarm dari Bagian ini memotong urutan tersebut di tengah malam: bila stub alarm_kegagalan sudah diganti notifikasi nyata, operator bangun mengetahui partisi mana yang gagal bahkan sebelum ritual pagi dimulai.

Pola logging dan metrik untuk pipeline dibahas pada chapter Monitoring Workflow dan Pipeline Logging, dan panduan UI resmi ada di dokumentasi Apache Airflow .