Caching

Setiap pukul 05:30 pipeline TokoKita menghitung ulang ringkasan GMV dari 300 ribu baris pesanan — padahal sebagian besar pagi, berkas sumber tidak berubah sejak kemarin. Chapter Spark Performance memperlihatkan cache() menyelamatkan job dari perhitungan ulang di cluster; prinsip yang sama berlaku untuk pipeline Python biasa, tetapi disiplinnya lebih ketat: cache harus punya kunci yang benar, masa berlaku yang jujur, dan perlindungan saat banyak permintaan datang serentak. Bab ini membangun cache disk sederhana dari nol supaya ketiga disiplin itu terlihat dari angka — dengan disiplin pengukuran before-after dari chapter Mengukur Performa Pipeline.

Kunci dari Hash Input #

Cache paling berbahaya adalah cache yang menjawab pertanyaan lama atas data baru. Solusinya: identitas hasil ditentukan sepenuhnya oleh input. Sidik jari SHA-256 atas byte berkas sumber menjadi nama berkas cache; satu byte berubah, kunci berubah, dan entri lama tidak akan pernah terbaca lagi. Invalidasi terjadi oleh konstruksi, bukan oleh jadwal TTL yang bisa telat:

def langkah_mahal(path_sumber: Path) -> pd.DataFrame:
    """Tahap pipeline: baca raw, validasi, agregasi GMV per kota."""
    df = pd.read_csv(path_sumber)
    df = df[df["qty"] > 0]
    return (df.assign(total=df["qty"] * df["harga"])
              .groupby("kota", as_index=False)["total"].sum()
              .sort_values("kota").reset_index(drop=True))


def hash_input(paths: list[Path]) -> str:
    """Sidik jari seluruh input: berubah satu byte pun, kunci berubah."""
    h = hashlib.sha256()
    for p in sorted(paths):
        h.update(p.read_bytes())
    return h.hexdigest()[:16]


def jalankan_dengan_cache(path_sumber: Path, label: str) -> float:
    """Jalankan tahap mahal dengan cache disk berbasis hash input."""
    kunci = hash_input([path_sumber])
    jalur_cache = CACHE / f"ringkasan-gmv-{kunci}.parquet"
    t0 = time.perf_counter()
    if jalur_cache.exists():
        hasil = pd.read_parquet(jalur_cache)
        status = "cache HIT "
    else:
        hasil = langkah_mahal(path_sumber)
        hasil.to_parquet(jalur_cache, index=False)
        status = "cache MISS"
    durasi = time.perf_counter() - t0
    print(f"{label}: {status} {durasi * 1000:8.1f} ms "
          f"(kunci={kunci}, {len(hasil)} baris hasil)")
    return durasi

Bagian driver kemudian menjalankan empat skenario: dua run atas input v1, satu baris pesanan baru ditambahkan ke sumber, lalu dua run atas input v2:

if __name__ == "__main__":
    print("=== Bagian 1: miss, hit, lalu invalidasi karena input berubah ===")
    sumber = tulis_sumber("pesanan-raw.csv", 300_000)
    jalankan_dengan_cache(sumber, "run 1 (input v1)")
    jalankan_dengan_cache(sumber, "run 2 (input v1, tak berubah)")

    # sumber berubah: satu hari data baru ditambahkan
    with open(sumber, "a", newline="") as f:
        tulis = csv.writer(f)
        tulis.writerow([999_999, "Jakarta", 7, 2, 250_000])
    jalankan_dengan_cache(sumber, "run 3 (input v2, satu baris baru)")
    jalankan_dengan_cache(sumber, "run 4 (input v2, tak berubah)")

Output yang Dihasilkan (bagian 1 dari 2):

=== Bagian 1: miss, hit, lalu invalidasi karena input berubah ===
run 1 (input v1): cache MISS    132.1 ms (kunci=b6f96f05c0d7bcb1, 8 baris hasil)
run 2 (input v1, tak berubah): cache HIT       7.1 ms (kunci=b6f96f05c0d7bcb1, 8 baris hasil)
run 3 (input v2, satu baris baru): cache MISS    110.0 ms (kunci=141e6906b79e3e75, 8 baris hasil)
run 4 (input v2, tak berubah): cache HIT       2.4 ms (kunci=141e6906b79e3e75, 8 baris hasil)

Mari kita telaah bagian-bagian penting kode di atas:

1. Baris 44-49 (hash_input): SHA-256 dihitung atas seluruh byte berkas, dengan daftar path diurutkan lebih dulu supaya urutan argumen tidak mengubah kunci; 16 digit pertama dipakai sebagai nama berkas.
2. Baris 54-55 (kunci dan jalur cache): nama berkas cache memuat kunci, sehingga dua versi input hidup berdampingan tanpa saling menimpa — run 3 tidak menimpa hasil run 1.
3. Baris 57-62 (cabang HIT/MISS): keberadaan berkas menentukan jalur eksekusi; HIT membaca parquet hasil, MISS menghitung lalu menulis.
4. Baris 77-79 (input berubah): satu baris pesanan Jakarta ditambahkan ke berkas sumber — cukup satu baris untuk mengganti kunci dari b6f9... menjadi 141e....
5. Baris 71-81 (urutan run): run 1 MISS, run 2 HIT 7,1 ms, run 3 MISS lagi setelah input berubah — invalidasi terjadi karena kunci berubah, bukan karena umur cache.

Cache Stampede dan Single-Flight #

Jebakan kedua muncul ketika cache baru saja kedaluwarsa atau belum terisi sementara empat permintaan datang serentak. Tanpa koordinasi, keempatnya menemukan cache kosong dan keempatnya menghitung penuh:

    print("=== Bagian 2: cache stampede, 4 permintaan serentak ===")
    hitung = [0]

    def versi_bebas():
        """Tanpa single-flight: tiap permintaan menghitung sendiri."""
        langkah_mahal(sumber)
        hitung[0] += 1

    utas = [threading.Thread(target=versi_bebas) for _ in range(4)]
    t0 = time.perf_counter()
    for u in utas:
        u.start()
    for u in utas:
        u.join()
    print(f"tanpa single-flight : komputasi mahal dijalankan {hitung[0]} kali "
          f"dalam {time.perf_counter() - t0:.2f} detik")

    hitung[0] = 0
    kunci_eksekusi = threading.Lock()
    cache_memori: dict[str, pd.DataFrame] = {}

    def versi_single_flight():
        """Single-flight: satu komputasi, sisanya menunggu hasilnya."""
        with kunci_eksekusi:
            if "gmv" not in cache_memori:
                cache_memori["gmv"] = langkah_mahal(sumber)
                hitung[0] += 1
        return cache_memori["gmv"]

    utas = [threading.Thread(target=versi_single_flight) for _ in range(4)]
    t0 = time.perf_counter()
    for u in utas:
        u.start()
    for u in utas:
        u.join()
    print(f"dengan single-flight: komputasi mahal dijalankan {hitung[0]} kali "
          f"dalam {time.perf_counter() - t0:.2f} detik")

Output yang Dihasilkan (bagian 2 dari 2):

=== Bagian 2: cache stampede, 4 permintaan serentak ===
tanpa single-flight : komputasi mahal dijalankan 4 kali dalam 0.32 detik
dengan single-flight: komputasi mahal dijalankan 1 kali dalam 0.11 detik

Mari kita telaah baris-baris kode di atas:

1. Baris 87-90 (versi_bebas): tanpa koordinasi, empat utas masing-masing menjalankan langkah_mahal penuh — empat kali kerja untuk satu jawaban.
2. Baris 102-103 (kunci_eksekusi): satu threading.Lock dan satu kamus memori menjadi gerbang tunggal seluruh permintaan.
3. Baris 105-111 (versi_single_flight): pemegang kunci pertama menghitung dan mengisi cache; tiga permintaan lain antre di with, lalu menemukan hasil sudah tersedia dan langsung membacanya.

Angkanya membuktikan bedanya: empat komputasi dalam 0,32 detik menjadi satu komputasi dalam 0,11 detik — tiga permintaan lain cukup menunggu jawaban pertama alih-alih menghitung sendiri. Pada langkah yang memakan 30 menit di production, pola inilah yang menentukan apakah cluster menghadapi empat job besar yang serentak atau satu job dan tiga pembaca hasil. Namun perhatikan pula bahwa single-flight menggeser biaya ke latensi: permintaan yang datang belakangan menunggu sampai komputasi pertama selesai, sehingga pola ini paling cocok untuk langkah yang benar-benar mahal dan hasil yang benar-benar dipakai bersama.

Single-Flight Lintas Proses #

threading.Lock hanya melindungi satu proses, padahal pipeline produksi sering berjalan sebagai beberapa proses terpisah — orchestrator menjalankan ulang pipeline lama tepat saat jadwal harian menyala. Di batas itu, kunci harus hidup di sistem operasi. Modul fcntl mengunci berkas sehingga proses mana pun yang ingin memegang kunci akan menunggu:

@contextmanager
def kunci_berkas(jalur_lock: Path):
    """Kunci eksklusif lintas proses di atas satu berkas."""
    with open(jalur_lock, "w") as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)

Pola pemakaiannya sama dengan jalankan_dengan_cache: kunci dulu, periksa cache di dalam kunci, baru hitung dan tulis. Berbeda dari threading.Lock, berkas kunci terlihat oleh semua proses yang mengarah ke direktori cache yang sama. Kedua modul tersebut terdokumentasi di dokumentasi hashlib dan dokumentasi fcntl .

Apa yang Tidak Layak Di-cache #

Paralelisme dan caching menghemat waktu, tetapi keduanya tetap menuntut satu sumber daya yang tidak bisa dibeli tunai: memori. Ketika dataset tidak lagi muat di RAM, pilihan desain berpindah ke ranah byte per objek — pokok bahasan chapter Memory Optimization.