Pipeline Performance

Volume pesanan toko online kita tumbuh sepuluh kali lipat dalam tiga bulan, dan durasi pipeline harian merambat dari 40 detik menjadi 6 menit — makin dekat ke tenggat laporan pukul 06:00. Keinginan pertama biasanya menebak: ganti pustaka, tambah thread, beli mesin lebih besar. Tanpa pengukuran, semua tebakan tersebut berpeluang memperbaiki bagian yang memang bukan masalah, sementara bottleneck sejati tetap utuh. Bab ini membahas cara memprofil durasi per stage, membaca letak bottleneck, lalu memperbaikinya dengan batching dan paralelisasi yang hasilnya terukur.

Menemukan Bottleneck: I/O, CPU, atau Serialisasi #

Pipeline batch umumnya menghabiskan waktu di tiga jenis bottleneck. Bottleneck I/O terjadi ketika stage menunggu disk atau jaringan — prosesor nyaris menganggur. Bottleneck CPU terjadi ketika satu stage memeras prosesor penuh, biasanya transformasi per baris dalam Python murni. Bottleneck serialisasi lebih tersembunyi: waktu habis untuk mengonversi data antar format atau memindahkan data antar proses. Ketiganya menuntut obat berbeda, sehingga dugaan harus digantikan oleh angka.

Jenis bottleneck	Gejala khas	Obat umum
I/O	stage menunggu disk atau jaringan, prosesor menganggur	batching, mengurangi jumlah commit, kompresi
CPU	satu stage memakan hampir seluruh durasi, prosesor penuh	vektorisasi, pustaka berbasis C, paralelisasi proses
Serialisasi	waktu habis untuk konversi antar format atau antar proses	mengurangi perpindahan data, format kolom

Profil Sederhana: Durasi per Stage #

Alat profil paling sederhana adalah time.perf_counter(): catat waktu sebelum dan sesudah setiap stage, lalu bagikan persentase terhadap total. Skrip berikut menyiapkan 100.000 baris pesanan dalam empat berkas partisi, memprofil pipeline versi pertama, lalu mencoba dua optimasi. Angka pastinya berubah-ubah antar mesin dan antar eksekusi — pola proporsinya yang stabil dan itulah yang kita baca.

python
"""Profil durasi per stage, temukan bottleneck, lalu optimasi dengan batching."""
import json
import sqlite3
import time
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path

DATA = Path("data")
DB = Path("toko_profil.db")
PARTISI = 4
BARIS_PER_PARTISI = 25000

PELANGGAN = ["Budi", "Siti", "Dewi", "Rina"]
PRODUK = [("Kemeja Batik", 240000), ("Kain Songket", 850000),
          ("Rok Tenun", 310000), ("Blouse Katun", 120000)]


def siapkan_data():
    DATA.mkdir(exist_ok=True)
    for p in range(PARTISI):
        baris = []
        for i in range(BARIS_PER_PARTISI):
            nama, harga = PRODUK[i % 4]
            baris.append(json.dumps({
                "pesanan_id": p * BARIS_PER_PARTISI + i + 1,
                "waktu": f"2026-05-1{p}T09:00:00",
                "pelanggan": PELANGGAN[i % 4], "produk": nama,
                "qty": (i % 3) + 1, "harga": harga, "diskon": 0}))
        (DATA / f"partisi_{p}.jsonl").write_text("\n".join(baris) + "\n")
    ukuran = sum(f.stat().st_size for f in DATA.glob("*.jsonl"))
    print(f"sumber: {PARTISI} berkas x {BARIS_PER_PARTISI} baris "
          f"({ukuran / 1_000_000:.1f} MB)")


def baca_satu(partisi):
    """Baca dan parsing satu berkas partisi (dipakai juga mode paralel)."""
    return [json.loads(baris)
            for baris in (DATA / f"partisi_{partisi}.jsonl").read_text().splitlines()]


def transform(mentah):
    return [(r["pesanan_id"], r["waktu"], r["pelanggan"], r["produk"],
             r["qty"], r["harga"], r["qty"] * r["harga"]) for r in mentah]


def buat_tabel(koneksi):
    koneksi.execute("DROP TABLE IF EXISTS pesanan")
    koneksi.execute("""CREATE TABLE pesanan (
        pesanan_id INTEGER PRIMARY KEY, waktu TEXT NOT NULL,
        pelanggan TEXT NOT NULL, produk TEXT NOT NULL,
        qty INTEGER NOT NULL, harga INTEGER NOT NULL, total INTEGER NOT NULL)""")


siapkan_data()
koneksi = sqlite3.connect(DB)

print("=== Versi 1: profil per stage ===")
mulai = time.perf_counter()
mentah = []
for p in range(PARTISI):
    mentah.extend(baca_satu(p))
durasi_extract = time.perf_counter() - mulai

mulai = time.perf_counter()
baris = transform(mentah)
durasi_transform = time.perf_counter() - mulai

buat_tabel(koneksi)
mulai = time.perf_counter()
for r in baris:
    koneksi.execute("INSERT INTO pesanan VALUES (?, ?, ?, ?, ?, ?, ?)", r)
    koneksi.commit()
durasi_load = time.perf_counter() - mulai

total = durasi_extract + durasi_transform + durasi_load
print(f"  extract   : {durasi_extract:.3f} detik ({durasi_extract/total*100:.0f}%)")
print(f"  transform : {durasi_transform:.3f} detik ({durasi_transform/total*100:.0f}%)")
print(f"  load      : {durasi_load:.3f} detik ({durasi_load/total*100:.0f}%)")
print(f"  total     : {total:.3f} detik")

print("=== Versi 2: load dengan batching (executemany + commit per batch) ===")
buat_tabel(koneksi)
mulai = time.perf_counter()
UKURAN_BATCH = 5000
for awal in range(0, len(baris), UKURAN_BATCH):
    koneksi.executemany(
        "INSERT INTO pesanan VALUES (?, ?, ?, ?, ?, ?, ?)",
        baris[awal:awal + UKURAN_BATCH])
    koneksi.commit()
durasi_load_batch = time.perf_counter() - mulai
print(f"  load batch: {durasi_load_batch:.3f} detik "
      f"({durasi_load / durasi_load_batch:.0f}x lebih cepat)")

print("=== Eksperimen: paralelisasi extract per partisi (4 proses) ===")
mulai = time.perf_counter()
mentah = []
for p in range(PARTISI):
    mentah.extend(baca_satu(p))
serial = time.perf_counter() - mulai
mulai = time.perf_counter()
with ProcessPoolExecutor(max_workers=4) as pool:
    hasil = list(pool.map(baca_satu, range(PARTISI)))
paralel = time.perf_counter() - mulai
jumlah = sum(len(bagian) for bagian in hasil)
print(f"  serial  : {serial:.3f} detik")
print(f"  paralel : {paralel:.3f} detik (pembagian hasil {jumlah} baris via pickle)")
print(f"  catatan : paralel {'lebih cepat' if paralel < serial else 'lebih LAMBAT'}"
      f" di sini karena biaya serialisasi antar proses")

koneksi.close()

Output yang Dihasilkan:

text
sumber: 4 berkas x 25000 baris (14.0 MB)
=== Versi 1: profil per stage ===
  extract   : 0.254 detik (8%)
  transform : 0.033 detik (1%)
  load      : 3.031 detik (91%)
  total     : 3.318 detik
=== Versi 2: load dengan batching (executemany + commit per batch) ===
  load batch: 0.082 detik (37x lebih cepat)
=== Eksperimen: paralelisasi extract per partisi (4 proses) ===
  serial  : 0.238 detik
  paralel : 0.390 detik (pembagian hasil 100000 baris via pickle)
  catatan : paralel lebih LAMBAT di sini karena biaya serialisasi antar proses

Mari kita telaah baris-baris kode di atas:

Baris 57-79 (profil versi 1): setiap stage dibungkus pasangan time.perf_counter() sebelum dan sesudah; pembagian durasi terhadap total menghasilkan persentase yang langsung menunjukkan letak masalah — di sini load 91%.
Baris 69-73 (load versi 1): pola execute lalu commit per baris. Setiap commit memaksa SQLite menyimpan ke disk, sehingga 100.000 baris berarti 100.000 kali operasi tulis fisik — inilah bottleneck I/O yang sejati.
Baris 81-92 (load versi 2): executemany memuat satu batch 5.000 baris dalam satu pernyataan, dan commit hanya dilakukan sekali per batch — 20 commit alih-alih 100.000.
Baris 94-108 (eksperimen paralel): extract dijalankan serial, lalu diulang dengan ProcessPoolExecutor empat proses; hasil kedua mode dibandingkan langsung, bukan diasumsikan.
Baris 35-38 (baca_satu): satu fungsi pembaca dipakai baik mode serial maupun paralel, sehingga perbandingan durasi mengukur murni efek paralelisasi, bukan perbedaan kode.

Batching pada Stage Load #

Profil versi pertama menunjukkan 91% waktu habis di stage load, sehingga mempercepat extract atau transform tidak akan terasa. Batching menyerang akar masalahnya: menggabungkan banyak baris ke dalam satu pernyataan executemany dan mengurangi jumlah commit. Hasil pada mesin uji: 3,031 detik turun ke 0,082 detik atau 37 kali lebih cepat, dan total pipeline menyusut dari 3,3 detik menjadi sekitar 0,4 detik. Prinsip yang sama berlaku luas: baca berkas dengan buffer besar, kirim data lewat jaringan dalam potongan besar, dan pada ekosistem kolom, proses data per kelompok baris seperti RecordBatch di Apache Arrow .

Ketika Paralelisasi Malah Memperlambat #

Eksperimen terakhir di skrip tersebut adalah pelajaran penting. Memecah extract ke empat proses justru memperlambat: 0,390 detik dibanding 0,238 detik serial. Penyebabnya adalah biaya serialisasi — hasil 100.000 baris di proses pekerja harus dipickle, dikirim ke proses utama, lalu dibangun kembali, dan biaya tersebut melebihi keuntungan memproses empat berkas bersamaan. Paralelisasi memang menolong ketika tiap pekerjaan benar-benar mandiri, menunggu I/O jaringan, atau memproses tanpa memindahkan data besar lintas proses — tetapi keputusannya harus lahir dari pengukuran, bukan intuisi. Dokumentasi resmi time.perf_counter() ada di dokumentasi Python .

Pipeline kita kini cepat dan setiap keputusannya punya angka pendukung. Satu bahan terakhir menentukan apakah pipeline tersebut bisa dioperasikan orang lain pada pukul tiga pagi: dokumentasi — pokok bahasan Chapter Dokumentasi Pipeline.