Logging dan Penanganan Error

Pipeline yang sudah dipasangi gerbang validasi kini punya cara menolak data bermasalah, tetapi pipeline yang sehat juga harus bisa menjelaskan apa yang terjadi ketika dijalankan. Bila pipeline berjalan tiap malam tanpa manusia di depan layar, deretan print() yang ditulis asal tidak cukup: tidak ada tingkat kepentingan, tidak ada penamaan sumber pesan, dan pesan bercampur dengan output program. Modul standar Python bernama logging menyelesaikan catatan perjalanan ini secara rapi, sementara blok try/except menjaga agar satu sumber yang gagal tidak menghentikan seluruh proses. Chapter ini memasang keduanya pada pipeline Toko Berkah Abadi dan menguji tiga skenario kegagalan yang paling sering muncul.

Konfigurasi Logger yang Rapi #

Tiga keputusan konfigurasi menentukan kualitas log pipeline:

1. Logger bernama: logging.getLogger("pipeline") memberi identitas pada pesan, sehingga log pipeline dapat dibedakan dari log modul lain.
2. Format tanpa timestamp: Formatter("%(levelname)s:%(name)s:%(message)s") mencetak level, nama logger, dan isi pesan. Timestamp memang berguna di produksi, tetapi format tanpa waktu membuat output mudah dibandingkan antar eksekusi.
3. Handler ke stdout: StreamHandler(sys.stdout) mengirim log ke arah yang sama dengan output program. Pengaturan bawaan logging.basicConfig() justru mengirim ke stderr, sehingga log dan hasil terpisah tempat.

Setelah logger dikonfigurasi sekali, fungsi tahap apa pun bisa menulis log dengan log.info(...), log.warning(...), atau log.error(...) tanpa tahu detail format.

Skrip: Log untuk Setiap Skenario #

Script berikut mendefinisikan tiga fungsi berjaring pengaman — baca_csv_aman() untuk berkas hilang, tambah_total() untuk kolom wajib yang tidak ada, dan konversi_jumlah() untuk nilai yang tidak bisa dikonversi — lalu mengujinya satu per satu sebelum menjalankan jalur sehat.

"""Pipeline dengan module logging dan penanganan error per tahap."""
import io
import logging
import sys
from pathlib import Path

import pandas as pd

TEKS_CSV = """id_pesanan,cabang,jumlah,harga
P001,Jakarta,2,45000
P002,Bandung,3,18000
P003,Surabaya,1,62000
"""

TEKS_KOTOR = """id_pesanan,jumlah
P001,2
P002,dua
"""


def siapkan_logger() -> logging.Logger:
    """Logger dengan format tanpa timestamp agar output deterministik."""
    logger = logging.getLogger("pipeline")
    logger.setLevel(logging.INFO)
    logger.handlers.clear()
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
    logger.addHandler(handler)
    return logger


log = siapkan_logger()


def baca_csv_aman(path: Path) -> pd.DataFrame:
    """Tahap extract dengan jaring pengaman FileNotFoundError."""
    log.info("extract : membaca %s", path)
    try:
        return pd.read_csv(path)
    except FileNotFoundError as e:
        log.error("extract : gagal, %s", e)
        return pd.DataFrame()


def tambah_total(df: pd.DataFrame) -> pd.DataFrame:
    """Tahap transform: kolom wajib hilang tidak menghentikan pipeline."""
    df = df.copy()
    try:
        df["total"] = df["jumlah"] * df["harga"]
        log.info("transform: kolom total dibuat (%d baris)", len(df))
    except KeyError as e:
        log.warning("transform: kolom %s tidak ada, total tidak dibuat", e)
    return df


def konversi_jumlah(df: pd.DataFrame) -> pd.Series:
    """Konversi kolom jumlah dengan jaring ValueError."""
    try:
        return df["jumlah"].astype(int)
    except ValueError as e:
        log.error("konversi : %s, pakai errors='coerce'", e)
        return pd.to_numeric(df["jumlah"], errors="coerce")


if __name__ == "__main__":
    print("== Sumber tidak ada ==")
    kosong = baca_csv_aman(Path("data_hilang.csv"))
    print(f"DataFrame kosong: {kosong.empty}")
    print()

    print("== Kolom wajib tidak ada ==")
    hasil = tambah_total(kosong)
    print(f"kolom setelah transform: {list(hasil.columns)}")
    print()

    print("== Jumlah tidak bisa dikonversi ==")
    kotor = pd.read_csv(io.StringIO(TEKS_KOTOR))
    jumlah_bersih = konversi_jumlah(kotor)
    print(f"hasil konversi: {jumlah_bersih.tolist()}")
    print()

    print("== Jalur sehat ==")
    Path("pesanan.csv").write_text(TEKS_CSV, encoding="utf-8")
    try:
        df = baca_csv_aman(Path("pesanan.csv"))
        df = tambah_total(df)
        ringkasan = (
            df.groupby("cabang", as_index=False)["total"].sum()
            .sort_values("total", ascending=False)
            .reset_index(drop=True)
        )
        print(ringkasan.to_string(index=False))
    finally:
        Path("pesanan.csv").unlink(missing_ok=True)

Output yang Dihasilkan:

== Sumber tidak ada ==
INFO:pipeline:extract : membaca data_hilang.csv
ERROR:pipeline:extract : gagal, [Errno 2] No such file or directory: 'data_hilang.csv'
DataFrame kosong: True

== Kolom wajib tidak ada ==
WARNING:pipeline:transform: kolom 'jumlah' tidak ada, total tidak dibuat
kolom setelah transform: []

== Jumlah tidak bisa dikonversi ==
ERROR:pipeline:konversi : invalid literal for int() with base 10: 'dua', pakai errors='coerce'
hasil konversi: [2.0, nan]

== Jalur sehat ==
INFO:pipeline:extract : membaca pesanan.csv
INFO:pipeline:transform: kolom total dibuat (3 baris)
  cabang  total
 Jakarta  90000
Surabaya  62000
 Bandung  54000

Mari kita bedah perilakunya per skenario:

1. Berkas tidak ada: pd.read_csv("data_hilang.csv") melempar FileNotFoundError dengan pesan [Errno 2] No such file or directory. Fungsi baca_csv_aman() mencatatnya sebagai ERROR lalu mengembalikan DataFrame kosong; pipeline berlanjut alih-alih berhenti dengan traceback.
2. Kolom wajib tidak ada: tambah_total() mencoba df["jumlah"] * df["harga"] pada DataFrame kosong. pandas melempar KeyError 'jumlah'; fungsi mencatat WARNING dan mengembalikan DataFrame tanpa kolom total, sehingga tahap berikutnya masih punya data untuk bekerja.
3. Nilai tak terkonversi: kolom jumlah berisi teks dua, sehingga astype(int) melempar ValueError: invalid literal for int() with base 10: 'dua'. Fungsi konversi_jumlah() mencatat ERROR lalu beralih ke pd.to_numeric(errors="coerce"), yang mengubah nilai buruk menjadi NaN dan angka valid tetap utuh.
4. Jalur sehat: sumber pesanan.csv ditulis, dibaca, dan ditransform dengan sukses; dua log INFO mencatat kedua tahap, dan ringkasan omzet tercetak seperti biasa. Blok finally menghapus berkas sementara agar tidak ada sisa.

Perhatikan bahwa seluruh pesan log muncul di stdout berdampingan dengan output program, persis seperti yang direncanakan lewat handler.

Memilih Level Log #

Level bukan formalitas; level menentukan bagaimana pembaca log menindaklanjuti pesan.

Dengan log yang rapi dan jaring pengaman per tahap, pipeline sudah layak dipercaya untuk berjalan sendiri. Tahap akhir yang belum kita bahas adalah menulis hasil ke tujuan yang sebenarnya — berkas dan database — dan itulah fokus chapter Ekspor Data Hasil Transformasi.

Referensi resmi: pandas.DataFrame.astype , pandas.to_numeric , dan pandas.read_csv .