Pipeline Logging

Chapter Logging di section Python untuk Data Engineering sudah membahas level, formatter, dan logger per modul. Skala pipeline menuntut satu lapisan desain lagi: satu run pipeline terdiri dari banyak stage yang dieksekusi terpisah, dan saat hasil pagi hari salah, kita perlu menjawab “run yang mana, stage yang mana, berapa lama, berapa baris”. Log per baris teks tidak bisa dijumlah maupun difilter per field. Chapter ini merancang logging pipeline dengan structured logging berformat JSON, run id sebagai pengikat semua kejadian, dan event baku untuk siklus hidup setiap stage.

Dari Log Teks ke Log Terstruktur #

Log "stage ekstraksi selesai 0.4 detik 45000 baris" enak dibaca manusia, tetapi mesin tidak bisa memisahkan durasi dari jumlah baris. Structured logging menyelesaikannya: setiap baris log adalah satu objek JSON dengan kolom tetap, sehingga pertanyaan seperti “tampilkan semua stage_error pada run_id tertentu” menjadi sekadar filter pada kolom, bukan parsing teks. Format JSON Lines juga langsung bisa diserap log store seperti Loki atau Elasticsearch tanpa regex ekstraksi, dan agregasi seperti “rata-rata durasi_ms stage pemuatan minggu ini” menjadi satu query biasa.

Tiga keputusan desain menentukan mutu log pipeline:

Satu run id untuk semua baris. Nilai run_id dihasilkan sekali saat pipeline dimulai, lalu disisipkan ke setiap baris log; semua kejadian satu eksekusi bisa dikumpulkan kembali dengan satu filter.
Event baku, bukan teks bebas. Kolom event memakai kosakata tetap (run_start, stage_end) sehingga dashboards bisa menghitung frekuensinya.
Angka sebagai angka. durasi_ms bertipe number, bukan string, agar bisa dihitung rata-rata dan persentilnya.

Skema Event dan Level Pipeline #

Tabel berikut memetakan event siklus hidup pipeline ke level logging yang tepat:

Event	Level	Kapan dicatat
`run_start`	INFO	pipeline dimulai; menyertakan jumlah input
`stage_start`	INFO	satu stage mulai dieksekusi
`stage_end`	INFO	stage selesai; menyertakan `durasi_ms`
`stage_error`	ERROR	stage mengangkat exception; run berhenti
`row_skipped` / `row_error`	WARNING	satu baris bermasalah, run tetap lanjut
`run_end`	INFO	seluruh run selesai; menyertakan hasil akhir

Pembagian ini penting: ERROR berarti run gagal dan pager berbunyi, sedangkan WARNING berarti data tercatat tidak sempurna namun hasil akhir masih sah. Bila keduanya dicampur, tim operasi berhenti memercayai notifikasi. Level DEBUG kita simpan untuk detail internal seperti isi query dan parameter koneksi, yang hanya diaktifkan saat menyelidiki masalah.

Implementasi Pipeline Mini Tiga Stage #

Pipeline mini berikut membaca baris transaksi mentah, memvalidasi bentuknya, menghitung nilai pesanan, lalu menulis hasil ke berkas JSONL. Semua kejadian mengalir lewat satu helper catat() dan satu pembungkus jalankan_stage(); handler, formatter, dan level yang dipakai adalah API standar modul logging Python yang dibahas di https://docs.python.org/3/library/logging.html :

python
"""Pipeline mini 3 tahap dengan structured logging JSON, run_id, dan event stage."""

import json
import logging
import time
import uuid

RUN_ID = uuid.uuid4().hex[:12]
logger = logging.getLogger("pipeline_pesanan")

BARIS_MENTAH = ["TRX-001,2,24000", "TRX-002,1,15000", "TRX-003,tiga,27000",
                "TRX-004,4", "TRX-005,2,18000"]


class FormatterJson(logging.Formatter):
    """Ubah LogRecord menjadi satu baris JSON."""

    def format(self, record):
        payload = {"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
                   "level": record.levelname, "logger": record.name,
                   "run_id": RUN_ID, "pesan": record.getMessage()}
        ekstra = getattr(record, "ekstra", None)
        if ekstra:
            payload.update(ekstra)
        return json.dumps(payload, ensure_ascii=False)


def siapkan_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(FormatterJson())
    root = logging.getLogger()
    root.addHandler(handler)
    root.setLevel(logging.INFO)


def catat(tingkat, pesan, **kolom):
    """Catat satu event dengan kolom terstruktur."""
    logger.log(tingkat, pesan, extra={"ekstra": kolom})


def jalankan_stage(nama, fungsi, *args):
    """Bungkus satu stage dengan event start/end/error plus durasi."""
    catat(logging.INFO, "stage dimulai", event="stage_start", stage=nama)
    mulai = time.monotonic()
    try:
        hasil = fungsi(*args)
    except Exception as galat:
        catat(logging.ERROR, "stage gagal", event="stage_error", stage=nama,
              error_type=type(galat).__name__,
              durasi_ms=round((time.monotonic() - mulai) * 1000, 1))
        raise
    catat(logging.INFO, "stage selesai", event="stage_end", stage=nama,
          durasi_ms=round((time.monotonic() - mulai) * 1000, 1))
    return hasil


def ekstraksi(baris_mentah):
    baris_valid = []
    for nomor, baris in enumerate(baris_mentah, start=1):
        kolom = baris.split(",")
        if len(kolom) != 3:
            catat(logging.WARNING, "baris dilewati", event="row_skipped",
                  stage="ekstraksi", baris=nomor, alasan="jumlah kolom salah")
            continue
        baris_valid.append(kolom)
    return baris_valid


def transformasi(baris_valid):
    hasil = []
    for id_trx, qty, harga in baris_valid:
        try:
            hasil.append({"id": id_trx, "nilai": int(qty) * int(harga)})
        except ValueError:
            catat(logging.WARNING, "baris gagal dikonversi", event="row_error",
                  stage="transformasi", id_pesanan=id_trx,
                  alasan="qty bukan angka")
    return hasil


def pemuatan(hasil):
    with open("hasil_pesanan.jsonl", "w", encoding="utf-8") as f:
        for baris in hasil:
            f.write(json.dumps(baris) + "\n")
    return len(hasil)


if __name__ == "__main__":
    siapkan_logging()
    catat(logging.INFO, "run dimulai", event="run_start",
          pipeline="etl_pesanan_harian", jumlah_input=len(BARIS_MENTAH))
    baris_valid = jalankan_stage("ekstraksi", ekstraksi, BARIS_MENTAH)
    hasil = jalankan_stage("transformasi", transformasi, baris_valid)
    jumlah = jalankan_stage("pemuatan", pemuatan, hasil)
    catat(logging.INFO, "run selesai", event="run_end",
          pipeline="etl_pesanan_harian", baris_dimuat=jumlah)

Output yang Dihasilkan:

text
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "run dimulai", "event": "run_start", "pipeline": "etl_pesanan_harian", "jumlah_input": 5}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "stage dimulai", "event": "stage_start", "stage": "ekstraksi"}
{"ts": "2026-10-16T14:40:43", "level": "WARNING", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "baris dilewati", "event": "row_skipped", "stage": "ekstraksi", "baris": 4, "alasan": "jumlah kolom salah"}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "stage selesai", "event": "stage_end", "stage": "ekstraksi", "durasi_ms": 0.0}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "stage dimulai", "event": "stage_start", "stage": "transformasi"}
{"ts": "2026-10-16T14:40:43", "level": "WARNING", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "baris gagal dikonversi", "event": "row_error", "stage": "transformasi", "id_pesanan": "TRX-003", "alasan": "qty bukan angka"}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "stage selesai", "event": "stage_end", "stage": "transformasi", "durasi_ms": 0.0}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "stage dimulai", "event": "stage_start", "stage": "pemuatan"}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "stage selesai", "event": "stage_end", "stage": "pemuatan", "durasi_ms": 0.1}
{"ts": "2026-10-16T14:40:43", "level": "INFO", "logger": "pipeline_pesanan", "run_id": "215d9b28ed72", "pesan": "run selesai", "event": "run_end", "pipeline": "etl_pesanan_harian", "baris_dimuat": 3}

Nilai run_id, kolom ts, dan durasi_ms berbeda pada setiap eksekusi karena semuanya dihasilkan saat runtime; sisanya identik. Baca output tersebut seperti kronologi satu shift: run_start membuka run, tiga pasang stage_start/stage_end membingkai tiap stage, dua WARNING menandai baris yang tidak ikut diproses, dan run_end menutup dengan hasil akhir baris_dimuat: 3. Mari kita telaah bagian-bagian kuncinya:

FormatterJson.format(): membangun kamus kolom tetap lalu json.dumps menjadi satu baris; atribut ekstra pada LogRecord membawa kolom tambahan yang dikirim helper catat().
catat(tingkat, pesan, **kolom): satu pintu untuk semua event; pemanggil tidak perlu menyusun extra secara manual sehingga nama kolom konsisten.
jalankan_stage(): membungkus stage mana pun dengan stage_start, stage_end, atau stage_error; time.monotonic() dipakai karena jam monoton tidak terpengaruh perubahan waktu sistem.
row_skipped untuk TRX-004 dan row_error untuk TRX-003: keduanya WARNING — run tetap selesai dengan baris_dimuat: 3, dan keputusan “lanjut atau berhenti” terlihat eksplisit di log.

Apa yang Tidak Boleh Masuk Log #

Prinsip praktisnya: log menjawab pertanyaan tentang proses (berapa baris, berapa lama, gagal di mana), bukan tentang isi data. Bila kita butuh sampel baris bermasalah untuk investigasi, tulis baris tersebut ke tabel karantina dengan kontrol akses yang setara tabel produksi, lalu catat pointer-nya di log. Sebagai uji kesehatan, setiap jenis event harus bisa menjawab pertanyaan yang jelas; event yang tidak pernah dibaca siapa pun hanya menaikkan biaya penyimpanan tanpa menambah wawasan.

Log terstruktur menghasilkan data yang siap dihitung. Dari baris-baris JSON di atas, kita bisa menghitung success rate, volume baris, dan durasi per stage untuk banyak run sekaligus — dan dari angka-angka itulah sistem alert dibangun, materi chapter Metrics dan Alerting.