Mengukur Performa Pipeline

Pipeline harian TokoKita selesai dalam 40 detik atau 6 menit — selisihnya menentukan apakah laporan pukul 06:00 siap atau terlambat. Sayangnya, kesan “terasa lebih cepat” bukanlah data; tanpa angka, setiap optimasi hanyalah tebakan yang bisa memperbaiki bagian yang memang bukan masalah. Chapter Pipeline Performance telah memprofil durasi per stage dengan time.perf_counter(); bab ini menambahkan lapisan metodologinya: metrik apa yang dicatat, jebakan apa yang membiaskan angka, dan bagaimana disiplin baseline membuat setiap klaim perbaikan bisa dipertanggungjawabkan.

Tiga Metrik Dasar Pipeline #

Sebelum menulis satu baris kode instrumen, kita harus sepakat lebih dulu apa yang diukur. Tiga metrik berikut menutup hampir semua kebutuhan profiling pipeline batch:

Throughput cocok untuk menilai pipeline batch secara keseluruhan, sedangkan latency persentil lebih relevan untuk sistem interaktif — query dasbor dengan p50 10 milidetik tetapi p95 2 detik masih terasa lambat bagi pengguna yang kebetulan mengalami yang terlama.

Instrumen: Pipeline Tiga Stage #

Kita pakai pipeline dengan pola yang sudah dikenal: 60 ribu pesanan TokoKita dibaca dari JSONL (extract), diubah menjadi tuple (transform), lalu dimuat ke SQLite (load). Fungsi jalankan membungkus tiap stage dengan pasangan time.perf_counter() dan mengumpulkan durasi per operasi di stage load; fungsi profil membuang satu run pemanasan lalu mengambil median:

def jalankan(mode_load, ukur_memori=False):
    """Satu run pipeline utuh; kembalikan durasi tiap stage + metrik tambahan."""
    con = sqlite3.connect(DB)
    buat_tabel(con)
    metrik = {}
    if ukur_memori:
        tracemalloc.start()
    t0 = time.perf_counter()
    mentah = stage_extract()
    d_extract = time.perf_counter() - t0
    if ukur_memori:
        tracemalloc.reset_peak()
    t0 = time.perf_counter()
    baris = stage_transform(mentah)
    del mentah
    d_transform = time.perf_counter() - t0
    if ukur_memori:
        tracemalloc.reset_peak()
    t0 = time.perf_counter()
    latensi = stage_load(con, baris, mode_load)
    d_load = time.perf_counter() - t0
    if ukur_memori:
        metrik["memori_puncak_mb"] = tracemalloc.get_traced_memory()[1] / 1e6
        tracemalloc.stop()
    latensi.sort()
    metrik["p50_op_ms"] = latensi[len(latensi) // 2] * 1000
    metrik["p95_op_ms"] = latensi[min(int(len(latensi) * 0.95), len(latensi) - 1)] * 1000
    con.close()
    return d_extract, d_transform, d_load, metrik


def profil(mode_load, n_run=3, ukur_memori=False):
    """Pemanasan 1 run (dibuang) + median n_run per stage."""
    jalankan(mode_load, ukur_memori)
    hasil = [[] for _ in range(3)]
    metrik = None
    for _ in range(n_run):
        d1, d2, d3, metrik = jalankan(mode_load, ukur_memori)
        hasil[0].append(d1); hasil[1].append(d2); hasil[2].append(d3)
    return [statistics.median(h) for h in hasil], metrik, hasil

Output yang Dihasilkan (baseline: load satu baris satu commit):

=== Baseline: load satu baris satu commit ===
stage       durasi (s)  porsi     throughput
extract          0.124     6%    482,860 b/s
transform        0.026     1%  2,315,292 b/s
load             1.829    92%      32,806 b/s
TOTAL            1.979
latency per operasi : p50 0.028 ms | p95 0.038 ms
sebaran 3 run        : min 1.964 s, maks 1.980 s

Mari kita telaah bagian-bagian penting kode di atas:

1. Baris 3-4 (tabel per run): setiap run membangun tabel dari nol lewat buat_tabel, sehingga semua run mengerjakan beban yang persis sama — syarat mutlak untuk perbandingan yang adil.
2. Baris 7-8 (tracemalloc opsional): pengukuran memori diaktifkan lewat parameter, tidak permanen — alasannya dibuktikan pada jebakan pertama di bawah.
3. Baris 15 dan 23 (reset_peak): puncak memori dihitung ulang per stage; tanpa ini, puncak stage extract menutupi profil memori stage berikutnya.
4. Baris 27-31 (persentil): daftar latensi berisi durasi setiap operasi INSERT plus commit; setelah diurutkan, elemen tengah menjadi p50 dan elemen posisi 95% menjadi p95.
5. Baris 41 (run pemanasan): run pertama dibuang karena sering menyimpang — kebiasaan yang angkanya kita buktikan sebentar lagi.

Perhatikan kontras yang mengajarkan: latency per operasi hanya 0,028 milidetik, tetapi throughput stage load cuma 32 ribu baris/detik. Operasi tunggal memang cepat; yang memberatkannya adalah 60 ribu kali perulangan commit — throughput, bukan latency, yang menjadi masalah stage load.

Tiga Jebakan Pengukuran #

Angka baru bisa dipercaya bila tiga jebakan berikut disadari. Ketiganya dibuktikan dengan pengukuran nyata pada pipeline yang sama, bukan sekadar diperingatkan.

Jebakan 1 — efek pengamat. tracemalloc mencatat setiap alokasi Python, dan pencatatan itu sendiri tidak gratis. Ketika pipeline dijalankan dengan tracemalloc aktif, durasinya naik dari 1,979 detik menjadi 2,748 detik (+41%). Karena itu waktu dan memori diukur pada run terpisah: profil memori dibaca dari run khusus, angka waktu dari run bersih.

Jebakan 2 — run dingin vs run hangat. Pipeline yang sama dijalankan ulang di proses Python baru (skrip pekerja terpisah; waktu dihitung sejak awal proses, termasuk import): hasilnya 1,993 detik versus median 1,979 detik pada proses hangat. Biaya tetapnya cuma 14 milidetik karena stack-nya ringan — import json, sqlite3 terukur 9 milidetik. Tetapi import pandas pada proses baru terukur 1.151 milidetik: pipeline terjadwal tiap lima menit dengan stack berat membayar lebih dari satu detik setiap run sebelum mengerjakan apa pun.

Jebakan 3 — noise antar run. Sebaran tiga run baseline di atas hanya 1,964-1,980 detik (±0,4%) karena mesin sedang sepi. Pada percobaan lain di mesin yang sama ketika proses lain sibuk, sebaran itu pernah melebar sampai ±20%. Konsekuensinya langsung: klaim perbaikan 15% pada mesin sesibuk itu bisa jadi murni kebisingan penjadwalan.

Baseline JSON dan Disiplin Sebelum-Sesudah #

Profil baru berguna bila bisa dibandingkan dari waktu ke waktu. Polanya: simpan metrik baseline ke berkas JSON, ubah satu hal, ukur ulang dengan prosedur yang sama, lalu vonis dengan pita noise sebagai ambang keputusan:

BASELINE.write_text(json.dumps({
    "versi": "satu-per-satu",
    "total_detik": round(total_base, 3),
    "stage": {"extract": round(durasi_base[0], 3),
              "transform": round(durasi_base[1], 3),
              "load": round(durasi_base[2], 3)},
}, indent=2))
durasi_baru, metrik_baru, _ = profil("batch")
total_baru = cetak_laporan("Kandidat: load executemany batch 5000",
                           durasi_baru, metrik_baru)
delta = (total_baru - total_base) / total_base * 100
pita = (max(sebaran) - min(sebaran)) / 2 / total_base * 100

Output yang Dihasilkan:

versi                  total (s)  load (s)
satu-per-satu              1.979     1.829
batch 5000                 0.195     0.047
delta total: -90.2% (pita noise ±0.4%) -> PERBAIKAN NYATA

Kandidat perbaikan ini sengaja memakai temuan lama chapter Pipeline Performance — executemany dengan commit per batch — supaya fokus kita tetap pada prosedurnya: satu perubahan, satu pengukuran, satu vonis. Delta -90,2% jauh melampaui pita noise ±0,4%, sehingga klaim perbaikannya sah. Setelah perbaikan, profil berubah bentuk: porsi load turun dari 92% ke 24% dan extract kini menjadi stage terbesar (62%) — profil baru inilah titik awal investigasi berikutnya. Referensi resmi keduanya ada di dokumentasi tracemalloc dan dokumentasi time .

Metrik level pipeline kini terukur dan jujur terhadap noise. Namun ketika yang lambat bukan pipeline kita sendiri, melainkan sebuah query di dalam warehouse, kita butuh mikroskop yang berbeda: rencana eksekusi — pokok bahasan Chapter Query Profiling.