Pipeline kita kini tahu memproses satu tanggal dengan aman, tetapi belum ada yang membangunkannya setiap malam. Penjadwalan menjawab dua pertanyaan yang berbeda: kapan pipeline harus bangun, dan tanggal mana yang harus dikerjakan saat bangun. Pertanyaan pertama kita jawab dengan ekspresi cron; pertanyaan kedua dengan fungsi due_runs yang membaca watermark. Kedua logika kita tulis sebagai fungsi murni terhadap waktu sehingga dapat diuji dengan jam palsu tanpa menunggu detik sungguhan berlalu.

Ekspresi Cron Lima Bidang #

Ekspresi 30 2 * * * pada config/pipeline.yaml berarti pukul 02:30 setiap hari. Lima bidangnya secara berurutan:

BidangNilai contohArti
minute30menit ke-30
hour2jam 02 pagi
dom*setiap tanggal dalam bulan
month*setiap bulan
dow*setiap hari dalam minggu
python src/scheduler.py — parse_cron dan next_run (dipangkas) 
def _parse_bidang(token: str, minimum: int, maksimum: int) -> frozenset[int]:
    """Parse satu bidang cron: '*', '*/n', 'a', 'a-b', 'a-b/n', daftar."""
    nilai: set[int] = set()
    for bagian in token.split(","):
        langkah = 1
        if "/" in bagian:
            bagian, _, langkah_teks = bagian.partition("/")
            langkah = int(langkah_teks)
        if bagian == "*":
            awal, akhir = minimum, maksimum
        elif "-" in bagian:
            awal, akhir = int(bagian.partition("-")[0]), int(bagian.partition("-")[2])
        else:
            awal = akhir = int(bagian)
        nilai.update(range(awal, akhir + 1, langkah))
    return frozenset(nilai)


def next_run(ekspresi: str, setelah: datetime) -> datetime:
    """Waktu jalankan berikutnya yang > `setelah` (eksklusif)."""
    elemen = parse_cron(ekspresi)
    kandidat = setelah.replace(second=0, microsecond=0) + timedelta(minutes=1)
    for _ in range(366 * 24 * 60):        # batas pencarian satu tahun
        if (_hari_cocok(elemen, kandidat)
                and kandidat.hour in elemen["hour"]
                and kandidat.minute in elemen["minute"]):
            return kandidat
        kandidat += timedelta(minutes=1)
    raise ValueError(f"tidak ada jadwal dalam satu tahun: {ekspresi!r}")

Parser menerima *, */n, a, a-b, dan daftar; nama hari tidak didukung, sedangkan 0 dan 7 sama-sama berarti Minggu. Fungsi _hari_cocok menerapkan aturan OR cron klasik: bila dom dan dow sama-sama dibatasi, cukup salah satu yang cocok, sehingga 0 0 1 * 0 berarti “tanggal 1 atau Minggu, tengah malam”. Hasil nyatanya pada November 2026:

Output yang Dihasilkan:

text 
== parse_cron('30 2 * * *') ==
{'minute': frozenset({30}), 'hour': frozenset({2}), 'dom': frozenset({1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31}), 'month': frozenset({1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12}), 'dow': frozenset({0, 1, 2, 3, 4, 5, 6})}

== next_run: tiga contoh ==
setelah 2026-11-05 00:10:00 -> 2026-11-05 02:30:00
setelah 2026-11-05 02:30:00 -> 2026-11-06 02:30:00
setelah 2026-11-05 03:00:00 -> 2026-11-06 02:30:00

== next_run dengan dom/dow (aturan OR) ==
setelah 2026-11-05 12:00 -> 2026-11-08 00:00:00

due_runs: Mengejar yang Tercecer #

Mengetahui jam bangun saja tidak cukup: mesin yang mati tiga hari bangun dengan tunggakan tiga partisi, dan jadwal harian hanya memberi satu kesempatan per malam. Fungsi berikut memutuskan tanggal mana yang wajib dikerjakan:

python src/scheduler.py — due_runs (dipangkas) 
def due_runs(state, saat, raw_dir, lookback_days=7, max_attempts=3) -> list[date]:
    """Tanggal partisi yang wajib diproses pada waktu `saat`."""
    hasil: list[date] = []
    hari_ini = saat.date()
    for offset in range(1, lookback_days + 1):
        tanggal = hari_ini - timedelta(days=offset)
        kunci = tanggal.isoformat()
        if not partition_path(raw_dir, tanggal).exists():
            continue                      # tidak ada partisinya: bukan pekerjaan kita
        if state.has_success(kunci):
            continue                      # sudah selesai
        if state.attempts(kunci) >= max_attempts:
            continue                      # terus gagal: butuh intervensi manual
        hasil.append(tanggal)
    hasil.sort()
    return hasil

Empat keputusan di dalamnya:

  1. Hari ini tidak pernah masuk daftar: partisi hari ini belum tentu selesai ditulis sumber, jadi hanya tanggal lampau yang dikejar.
  2. Partisi harus ada dan belum sukses: tanggal tanpa berkas bukan tanggung jawab pipeline, dan tanggal yang sudah sukses dilewati lewat watermark.
  3. Percobaan terbatas max_attempts: tanggal yang tiga kali gagal berhenti dicoba otomatis dan menunggu diperiksa manusia.

Skenario berikut dijalankan pada data proyek dengan jam palsu:

Output yang Dihasilkan:

text 
== due_runs, jam palsu 2026-11-05 02:30 (state masih kosong) ==
['2026-11-02', '2026-11-03', '2026-11-04']

== setelah 02 dan 03 sukses, jam 2026-11-06 02:30 ==
['2026-11-04']

== semua sukses: tidak ada yang jatuh tempo ==
[]

Loop Penjadwal #

run_loop merangkai kedua fungsi: tidur sampai next_run, bangun, minta due_runs, lalu panggil runner untuk tiap tanggal. Jam dan fungsi tidur disuntik sebagai parameter. Daemon produksi dijalankan dengan python3 -m src.scheduler dan tidak kita eksekusi di sini karena memang tidak berhenti; tests/test_scheduler.py menguji siklusnya dengan jam palsu:

python src/scheduler.py — run_loop (dipangkas) 
while max_cycles is None or siklus < max_cycles:
    sekarang = now_fn()
    berikutnya = next_run(cron, sekarang)
    jeda = (berikutnya - sekarang).total_seconds()
    if jeda > 0:
        log(f"[scheduler] menunggu {jeda:.0f} detik "
            f"sampai {berikutnya:%Y-%m-%d %H:%M}")
        sleep_fn(jeda)
    sekarang = now_fn()
    jadwal = due_runs(state, sekarang, raw_dir, lookback_days, max_attempts)
    if jadwal:
        log("[scheduler] partisi jatuh tempo: "
            + ", ".join(d.isoformat() for d in jadwal))
        for tanggal in jadwal:
            runner(tanggal)
    siklus += 1

Dengan jam palsu yang maju dari 00:10 ke 02:30:05, satu siklus menghasilkan:

Output yang Dihasilkan:

text 
[scheduler] menunggu 8400 detik sampai 2026-11-05 02:30
(tidur 8400 detik)
[scheduler] partisi jatuh tempo: 2026-11-02, 2026-11-03, 2026-11-04
runner dipanggil untuk: ['2026-11-02', '2026-11-03', '2026-11-04']

Cron dan systemd di Produksi #

Dua berkas di config/ menyiapkan penjadwalan level sistem operasi. Keduanya bisa diunduh dari bagian Berkas Sumber; isinya sebagai berikut:

text config/cron 
# /etc/cron.d/etl-tokokita
# Jadwal pipeline ETL harian TokoKita dengan cron sistem.
# Instalasi: salin ke /etc/cron.d/etl-tokokita, lalu chmod 644.
# Karakter % harus ditulis %% karena cron menafsirkan % sebagai pemisah baris.

SHELL=/bin/sh
CRON_TZ=Asia/Jakarta

30 2 * * * tokokita cd /srv/tokokita/etl && /usr/bin/python3 -m src.pipeline --date yesterday >> data/state/cron.log 2>&1
text config/etl-nightly.timer 
[Unit]
Description=Penjadwal pipeline ETL harian TokoKita

[Timer]
OnCalendar=*-*-* 02:30:00
Persistent=true
RandomizedDelaySec=300

[Install]
WantedBy=timers.target
text config/etl-nightly.service 
[Unit]
Description=Menjalankan pipeline ETL pesanan TokoKita
After=network-online.target

[Service]
Type=oneshot
User=tokokita
WorkingDirectory=/srv/tokokita/etl
ExecStart=/usr/bin/python3 -m src.pipeline --date yesterday

Ketiga berkas mengasumsikan proyek dipasang di /srv/tokokita/etl dengan pengguna tokokita. Baris --date yesterday memakai fitur parse_run_date: kata tersebut diubah menjadi tanggal sebelumnya. Pada systemd, Persistent=true menjalankan timer yang terlewat begitu mesin hidup kembali; pasangan ini sejalan dengan due_runs yang mengejar partisi terlewat, tema chapter Backfill dan Scheduling. Untuk belajar, cron lebih sederhana; untuk produksi, systemd memberi journalctl dan kebijakan restart tanpa mengubah kode pipeline.

Dengan penjadwal di tempatnya, satu masalah terakhir pada rangkaian ini: kegagalan itu sendiri, yang kita tangani pada Bagian 7.

Bacaan lanjut: crontab(5) , systemd.timer , dan systemd.service .