Chapter Scheduling di section Orkestrasi Workflow telah membahas cara orchestrator menghitung jadwal — cron, data interval, catchup — dan chapter Backfill membahas syarat menjalankan ulang periode lampau. Di production, dua hal itu belum cukup. Ketika scheduler mati sepanjang akhir pekan, kita perlu tahu window mana saja yang tertunda dan mana yang aman dikejar; ketika run yang sama dipicu dua kali, kita perlu penjaga yang menolak yang kedua; dan ketika seseorang memicu run manual pukul tiga pagi, kita perlu jejak audit yang menjelaskan siapa dan mengapa. Semua kebutuhan itu bermuara pada satu artefak: run ledger, pembukuan setiap eksekusi pipeline yang lengkap dan bisa di-query.

Run Ledger sebagai Sumber Kebenatan #

UI orchestrator yang dibahas chapter Monitoring Workflow nyaman untuk menyelidiki satu run, tetapi pertanyaan produksi sering berbentuk agregat: berapa window yang tertunda, kapan sukses terakhir, siapa yang memicu run manual. Run ledger menjawabnya karena setiap eksekusi menambah satu baris JSONL dengan kolom tetap: run_id, pipeline, window (periode data yang diproses), actual_start, status, durasi_menit, dan triggered_by — kolom terakhir inilah yang membedakan scheduler dari [email protected] (manual, alasan: ...). Ledger bersifat append-only: rekaman lama tidak pernah diubah, sehingga riwayat dapat dibandingkan antar waktu tanpa ragu.

Mendeteksi Window Terlewat #

Pipeline harian TokoKita berjalan pukul 02:00 untuk memproses window hari sebelumnya — model data interval yang dijelaskan di chapter Scheduling. Empat fungsi berikut membentuk pembantu keputusan di atas ledger tersebut:

python 
"""Run ledger produksi: deteksi window terlewat, guard concurrency, audit manual."""
import json
from datetime import datetime, timedelta
from pathlib import Path

AKAR = Path(__file__).parent
LEDGER = AKAR / "ledger_runs.jsonl"
SEKARANG = datetime(2026, 10, 15, 6, 0)  # dipaku agar demo deterministik
JAM_RUN = 2  # pipeline harian berjalan tiap hari pukul 02:00
RETENSI_SUMBER = 30  # hari; sumber menyimpan histori selama ini


def catat_run(run_id, window, status, mulai, durasi_menit, oleh, galat=None):
    """Satu baris JSONL ke ledger: sumber kebenaran riwayat run."""
    rekaman = {"run_id": run_id, "pipeline": "etl_pesanan_harian",
               "window": window.isoformat(), "status": status,
               "actual_start": mulai.isoformat(), "durasi_menit": durasi_menit,
               "triggered_by": oleh}
    if galat:
        rekaman["galat"] = galat
    with open(LEDGER, "a", encoding="utf-8") as f:
        f.write(json.dumps(rekaman) + "\n")


def baca_ledger():
    return [json.loads(baris) for baris
            in LEDGER.read_text(encoding="utf-8").splitlines() if baris]


def jendela_terlewat(sukses_terakhir, sekarang):
    """Window tanpa catatan SUKSES, dari hari setelah sukses terakhir."""
    hasil = []
    indeks = {r["window"]: r for r in baca_ledger()}
    kandidat = sukses_terakhir + timedelta(days=1)
    while kandidat + timedelta(days=1, hours=JAM_RUN) <= sekarang:  # jadwal 02:00
        rekaman = indeks.get(kandidat.isoformat())
        if rekaman is None:
            hasil.append((kandidat, "TANPA RUN"))
        elif rekaman["status"] != "success":
            hasil.append((kandidat, f"{rekaman['status'].upper()} ({rekaman['run_id']})"))
        kandidat += timedelta(days=1)
    return hasil


def boleh_jalankan(window):
    """Guard concurrency: tolak bila window yang sama masih berjalan."""
    for r in baca_ledger():
        if r["window"] == window.isoformat() and r["status"] == "running":
            mulai = datetime.fromisoformat(r["actual_start"])
            sudah = (SEKARANG - mulai).total_seconds() / 3600
            return False, (f"run {r['run_id']} masih RUNNING "
                           f"(mulai {mulai:%H:%M}, sudah {sudah:.0f} jam)")
    return True, "tidak ada run berjalan untuk window ini"


def cek_kalender(jam_usulan, rawat, jadwal_lain):
    """Hindari tabrakan: jendela rawat dan pipeline lain di jam yang sama."""
    bentrok = [f"jendela rawat {nama} {mulai:02d}:00-{akhir:02d}:00"
               for nama, (mulai, akhir) in rawat.items()
               if mulai <= jam_usulan < akhir]
    bentrok += [f"{jumlah} pipeline lain di jam {jam_usulan:02d}:00"
                for jam, jumlah in jadwal_lain.items()
                if jam == jam_usulan and jumlah > 0]
    return bentrok or ["bersih (tidak bentrok)"]

Mari kita telaah baris-baris kode di atas:

  1. Baris 32-41 (catat_run): satu pintu untuk semua penulisan — entah pemicunya scheduler atau manusia, bentuk rekamannya sama, termasuk kolom galat opsional untuk run yang gagal.
  2. Baris 49-61 (jendela_terlewat): kandidat bergerak hari demi hari setelah sukses terakhir; kondisi kandidat + 1 hari + 2 jam memastikan hanya window yang seharusnya sudah dieksekusi pukul 02:00 yang dihitung. Setiap window diklasifikasikan TANPA RUN, FAILED, atau RUNNING — tiga kondisi menuntut tindakan berbeda.
  3. Baris 64-72 (boleh_jalankan): sebelum memicu run untuk satu window, fungsi ini memeriksa apakah window yang sama masih berstatus running; pesan penolakannya menyertakan run id dan sudah berapa lama berjalan.
  4. Baris 75-83 (cek_kalender): usulan jam run diperiksa terhadap jendela rawat (backup, upgrade) dan pipeline lain yang menempati jam sama.

Driver demo menabung empat rekaman riwayat — dua sukses terakhir, satu gagal karena sumber kosong, dan satu yang masih berjalan — lalu menjalankan ketiga pemeriksaan di atas plus satu trigger manual. Hasilnya dicetak sebagai laporan satu layar:

Output yang Dihasilkan:

text 
jadwal   : harian 02:00, sukses terakhir window 2026-10-10, sekarang 2026-10-15 06:00:00 (dipaku)
window terlewat:
  2026-10-11 -> TANPA RUN
  2026-10-12 -> TANPA RUN
  2026-10-13 -> FAILED (r-841)
  2026-10-14 -> RUNNING (r-850)

keputusan catch-up:
  3 window layak backfill; tertua 4 hari <= retensi sumber 30 hari -> aman, urutkan dari terlama

guard concurrency + audit manual:
  2026-10-14 -> LEWATI: run r-850 masih RUNNING (mulai 02:00, sudah 4 jam)
  2026-10-11 -> JALANKAN dan catat audit (tidak ada run berjalan untuk window ini)

kalender produksi:
  usulan 00:00 -> jendela rawat backup warehouse 00:00-01:00; 3 pipeline lain di jam 00:00
  usulan 02:00 -> bersih (tidak bentrok)

Baca keluaran tersebut sebagai satu rapat pagi yang singkat. Empat window tertunda terbagi dua kelompok: tiga layak backfill karena masih dalam usia retensi sumber, sedangkan window 10-14 ditunda karena run lama masih berjalan — mengejarnya sekarang berarti dua proses menulis partisi yang sama. Trigger manual untuk window 10-11 lolos guard dan tercatat lengkap dengan pemicunya. Konsep catchup otomatis versus backfill eksplisit serta syarat idempotent dibahas tuntas di chapter Backfill; perilaku penjadwal di balik jam 02:00 ada di dokumentasi Apache Airflow .

Run ledger yang rapi adalah bahan bakar alami untuk pertanyaan berikutnya: seberapa sehat pipeline ini berminggu-minggu? Dari baris-baris ledger yang sama kita bisa menghitung success rate, durasi, dan kecenderungan kegagalan — lalu merutekan notifikasinya. Itu pekerjaan chapter Monitoring dan Alerting.