Bagian 7: Menangani Kegagalan

Pipeline yang berjalan setiap malam tanpa pengawasan pasti menemui hari buruknya: partisi yang tidak turun, berkas dengan kolom hilang, atau warehouse yang menolak beban. Kegagalan seperti ini bukan keadaan darurat yang panik, melainkan kejadian yang didesain sejak awal: setiap jenis galat punya nomor keluar sendiri, setiap baris yang ditolak punya berkas karantina, dan setiap kejadian gagal tercatat di log alert. Tiga mekanisme ini bekerja tanpa jaringan dan tanpa layanan luar, sehingga bisa dijalankan dan diuji sepenuhnya di proyek ini.

Kontrak Exit Code #

Kelas galat di src/failures.py mewarisi satu induk dan membawa nomor keluar masing-masing:

class PipelineError(Exception):
    """Galat bawaan pipeline; subclass mengubah exit_code."""
    exit_code = 1


class ConfigError(PipelineError):
    exit_code = 2


class ExtractError(PipelineError):
    exit_code = 3


class LoadError(PipelineError):
    exit_code = 5


class ValidationError(PipelineError):
    exit_code = 6


def exit_code_for(exc: BaseException) -> int:
    """Exit code untuk sembarang jenis galat."""
    if isinstance(exc, PipelineError):
        return exc.exit_code
    return 1

Angka lengkapnya sudah kita tetapkan pada halaman overview: 0 sukses atau lewati, 1 argumen atau galat tak terduga, 2 konfigurasi, 3 ekstraksi, 4 transformasi, 5 pemuatan, dan 6 validasi. Fungsi main() di src/pipeline.py menangkap PipelineError lalu mengembalikan angka tersebut, sehingga pemanggil otomatis tinggal memeriksa $?.

Karantina Baris yang Ditolak #

Baris yang gagal aturan Bagian 3 tidak dibuang. Fungsi berikut menuliskannya sebagai CSV dead-letter:

def quarantine_rows(rows, path, kolom=("baris", "order_id", "alasan", "nilai")):
    """Tulis baris gagal validasi ke berkas dead-letter CSV."""
    if not rows:
        return None
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as f:
        penulis = csv.DictWriter(f, fieldnames=list(kolom))
        penulis.writeheader()
        penulis.writerows({k: r.get(k, "") for k in kolom} for r in rows)
    return path

Pada data proyek, dua partisi menghasilkan karantina:

Output yang Dihasilkan:

== data/quarantine/pesanan_2026-11-03.csv ==
baris,order_id,alasan,nilai
8,ORD-2026-11-03-006,customer_name kosong,
11,ORD-2026-11-03-009,city kosong,

== data/quarantine/pesanan_2026-11-04.csv ==
baris,order_id,alasan,nilai
9,ORD-2026-11-04-008,quantity tidak valid,-1

Kolom baris menunjuk baris berkas sumber, kolom alasan menyebut aturan yang dilanggar, dan kolom nilai menyimpan teks mentahnya. Operator dapat memperbaiki sumber lalu menjalankan ulang dengan --force; karena pemuatan bersifat upsert, hasil perbaikan menggantikan hasil lama tanpa menduplikasi.

Eksekusi yang Gagal #

Mari kita jalankan pipeline untuk tanggal yang partisinya tidak ada. Ini bukan simulasi; perintah berikut benar-benar dijalankan pada proyek:

python3 -m src.pipeline --date 2026-11-05; echo "exit=$?"

Output yang Dihasilkan:

[galat] ExtractError (exit 3): partisi tidak ditemukan: data/raw/pesanan_2026-11-05.csv
[alert] status=disabled kanal=- (stub: tidak ada permintaan jaringan)
[selesai] run-2026-11-05-075018 status=failed durasi=0ms exit=3
exit=3

Perhatikan urutan kerjanya: galat dicatat ke state (attempt bertambah), manifest berstatus failed ditulis ke data/state/runs/, lalu baris [galat] dikirim ke STDERR sementara baris [selesai] tetap di STDOUT.

Alert Stub Tanpa Jaringan #

Alert diaktifkan dengan menyalin contoh konfigurasi: cp config/alert.example.yaml config/alert.yaml. Stub membaca daftar kanal, menyaring yang berminat pada event tersebut, lalu menuliskan satu baris ke data/state/alerts.log. Nilai url pada contoh adalah penampung YOUR_WEBHOOK_URL, sehingga stub tidak pernah mengirim permintaan apa pun:

Setelah contoh disalin, eksekusi yang sama menghasilkan:

Output yang Dihasilkan:

[galat] ExtractError (exit 3): partisi tidak ditemukan: data/raw/pesanan_2026-11-05.csv
[alert] status=stub kanal=tim-data,penjaga-jadwal (stub: tidak ada permintaan jaringan)
[selesai] run-2026-11-05-075019 status=failed durasi=1ms exit=3
exit=3

Output yang Dihasilkan:

== data/state/alerts.log ==
2026-10-17T07:50:19+00:00 event=failed kanal=penjaga-jadwal,tim-data {'run_date': '2026-11-05', 'jenis': 'ExtractError'}

Baris pada log memakai urutan kanal terurut, sedangkan baris [alert] di layar mengikuti urutan kanal pada berkas konfigurasi. Event validation hanya menjangkau kanal yang mendaftarkan minatnya, sehingga saluran yang berbeda bisa menerima kejadian yang berbeda.

Percobaan Berulang dan Blokir #

Setiap kegagalan menambah attempt pada state. Scheduler dari Bagian 6 membaca angka itu: tanggal dengan attempt di bawah max_attempts dijadwalkan ulang malam berikutnya, sedangkan tanggal yang tiga kali gagal berhenti dikejar dan menunggu diperiksa manusia. Kebijakan percobaan berulang ini dibahas tuntas pada chapter Retries dan Timeouts, pola karantina pada Menangani Kegagalan Ingestion, dan pemulihan dari kegagalan pada Pipeline Recovery.

Kombinasi exit code, karantina, dan alert membuat kegagalan menjadi kejadian yang terukur, bukan keadaan yang merusak. Data yang ditolak tetap utuh di data/quarantine, warehouse tetap konsisten karena transaksi dibatalkan, dan penjadwal tahu persis kapan harus mencoba lagi. Pada bagian berikutnya kita melengkapi sisi pengamatan: manifest per eksekusi dan laporan status.

Bacaan lanjut: sys.exit dan csv modul pada dokumentasi Python.