Mengatasi Data Skew

Enam puluh persen pesanan TokoKita berasal dari Jakarta — kabar baik untuk bisnis, kabar buruk untuk pipeline yang membagi kerja berdasarkan kota. Satu kunci raksasa membuat satu worker menampung sebagian besar baris sementara worker lain selesai lebih cepat lalu menganggur menunggu. Chapter Shuffle dan Skew memperlihatkan masalah ini di Spark dan obat salting-nya; bab ini membawa persoalan yang sama ke alat non-Spark — hash partition di multiprocessing dan GROUP BY di DuckDB. Pada akhirnya kita juga menemukan kasus yang jujur: engine yang satu ternyata tidak memerlukan obat sama sekali.

Diagnosa Dulu, Baru Obat #

Langkah pertama mengatasi skew bukan menulis salting, melainkan menghitung. Query lima kunci teratas menunjukkan bentuk sebaran dalam waktu singkat:

    print("=== Bagian 1: diagnosa dulu - top 5 kunci terbesar ===")
    con = duckdb.connect()
    con.execute("SET enable_progress_bar=false")
    for kota, jumlah in con.execute(f"""
            SELECT kota, count(*) AS jumlah FROM read_csv('{sumber}')
            GROUP BY kota ORDER BY jumlah DESC LIMIT 5
        """).fetchall():
        print(f"{kota:15s} {jumlah:7,} baris ({jumlah / N * 100:.0f}%)")
    total_kunci = con.execute(f"""
        SELECT count(DISTINCT kota) FROM read_csv('{sumber}')
    """).fetchone()[0]
    print(f"jumlah kunci unik: {total_kunci}")

Output yang Dihasilkan (bagian 1 dari 3):

=== Bagian 1: diagnosa dulu - top 5 kunci terbesar ===
Jakarta          60,000 baris (60%)
Medan             2,175 baris (2%)
Batam             2,170 baris (2%)
Tangerang         2,132 baris (2%)
Palembang         2,132 baris (2%)
jumlah kunci unik: 20

Baris Jakarta 60,000 (60%) adalah vonisnya: satu kunci menguasai mayoritas dari 100 ribu baris, sementara 19 kota lain berbagi sisanya masing-masing sekitar 2%. Sebaran seperti ini khas data e-commerce Indonesia, dan bentuk sebarannyalah yang menentukan pilihan obat.

Partition Timpang di Multiprocessing #

Model hash partition sederhana: setiap baris dikirim ke worker sesuai hash(kunci) % jumlah_worker. Dua helper berikut menjadi inti simulasinya:

def partisi(kota: str, jumlah: int = JUMLAH_WORKER) -> int:
    """Arahkan baris ke worker berdasarkan hash kunci - model hash partition."""
    return zlib.crc32(kota.encode()) % jumlah


def agregasi_pekerja(baris_pekerja: list[tuple[str, float]]) -> dict[str, float]:
    """Tahap 1 di worker: agregasi parsial kota -> total."""
    parsial: dict[str, float] = {}
    for kota, nilai in baris_pekerja:
        parsial[kota] = parsial.get(kota, 0.0) + nilai
    return parsial

Simulasi kemudian menyebarkan 100 ribu baris ke empat keranjang sesuai partisi, lalu menjalankan agregasi parsial di empat proses:

    print("=== Bagian 2: hash-partition per kota (sebelum salting) ===")
    keranjang: list[list[tuple[str, float]]] = [[] for _ in range(JUMLAH_WORKER)]
    for kota, nilai in baris:
        keranjang[partisi(kota)].append((kota, nilai))

    with ProcessPoolExecutor(max_workers=JUMLAH_WORKER) as pool:
        hasil_parsial = list(pool.map(agregasi_pekerja, keranjang))

    for w, (isi, parsial) in enumerate(zip(keranjang, hasil_parsial)):
        kunci_di_sini = len(parsial)
        print(f"worker {w}: {len(isi):7,} baris, {kunci_di_sini} kunci unik")
    puncak = max(len(k) for k in keranjang)
    rata = N / JUMLAH_WORKER
    print(f"rasio puncak/rata-rata: {puncak / rata:.1f}x "
          f"(worker terberat menahan seluruh pipeline)")

Output yang Dihasilkan (bagian 2 dari 3):

=== Bagian 2: hash-partition per kota (sebelum salting) ===
worker 0:  18,917 baris, 9 kunci unik
worker 1:  12,762 baris, 6 kunci unik
worker 2:  64,118 baris, 3 kunci unik
worker 3:   4,203 baris, 2 kunci unik
rasio puncak/rata-rata: 2.6x (worker terberat menahan seluruh pipeline)

Worker 2 menerima 64.118 baris — 2,6 kali rata-rata — karena hash menempatkan Jakarta bersama dua kota lain di keranjang yang sama, sementara worker 3 hanya menerima 4.203 baris. Durasi pipeline paralel ditentukan worker terberat, jadi tiga worker yang cepat hanya menunggu: empat core yang dipakai efektifnya seperti kurang dari dua core.

Salting dan Agregasi Dua Tahap #

Obatnya sejenis dengan yang dipakai Spark di chapter Shuffle dan Skew: pecah kunci panas dengan garam sehingga salinannya tersebar ke seluruh worker, lalu gabungkan kembali hasil parsialnya:

    print("=== Bagian 3: salting + agregasi dua tahap (setelah perbaikan) ===")
    keranjang_garam: list[list[tuple[str, float]]] = [
        [] for _ in range(JUMLAH_WORKER)]

    def partisi_bergaram(kota: str, garam: int) -> int:
        """Garam menggeser tujuan worker: salinan kunci yang sama tersebar
        ke seluruh worker (hash kota + garam, modulo jumlah worker)."""
        return (zlib.crc32(kota.encode()) + garam) % JUMLAH_WORKER

    for i, (kota, nilai) in enumerate(baris):
        keranjang_garam[partisi_bergaram(kota, i % JUMLAH_WORKER)].append(
            (kota, nilai))

    with ProcessPoolExecutor(max_workers=JUMLAH_WORKER) as pool:
        hasil_garam = list(pool.map(agregasi_pekerja, keranjang_garam))

    for w, isi in enumerate(keranjang_garam):
        print(f"worker {w}: {len(isi):7,} baris")
    puncak_garam = max(len(k) for k in keranjang_garam)
    print(f"rasio puncak/rata-rata: {puncak_garam / rata:.1f}x")

Tahap kedua di akhir skrip (dipangkas dari tampilan) menjumlahkan hasil parsial tiap worker per kota, lalu membandingkannya dengan agregasi langsung atas seluruh 100 ribu baris:

Output yang Dihasilkan (bagian 3 dari 3):

=== Bagian 3: salting + agregasi dua tahap (setelah perbaikan) ===
worker 0:  24,856 baris
worker 1:  25,060 baris
worker 2:  24,969 baris
worker 3:  25,115 baris
rasio puncak/rata-rata: 1.0x
total GMV Jakarta dua tahap : Rp22,668,681,000
hasil identik agregasi langsung: True (20 kota, 100,000 baris)

Mari kita telaah bagian-bagian penting kode di atas:

1. Baris 34-36 (partisi): zlib.crc32 mengubah nama kota menjadi angka, lalu modulo jumlah worker menentukan keranjang; kunci yang sama selalu mendarat di keranjang yang sama — itulah inti hash partition sekaligus sumber skew.
2. Baris 39-44 (agregasi_pekerja): worker menghitung total parsial per kota tanpa perlu mengetahui kota di worker lain.
3. Baris 91-94 (partisi_bergaram): garam 0-3 ditambahkan ke hash sebelum modulo, sehingga 60 ribu baris Jakarta menyebar ke empat keranjang, bukan menumpuk di satu.
4. Baris 96-98 (loop penyebaran): nomor baris i menjadi sumber garam i % 4; rasio puncak turun dari 2,6x ke 1,0x dan hasil dua tahap terbukti identik.

DuckDB Tidak Selalu Butuh Obat #

Apakah dua tahap juga mempercepat GROUP BY di DuckDB? Pengujian yang sama dijalankan pada 4 juta baris berpola 60% Jakarta — agregasi langsung versus dua tahap dengan garam pesanan_id % 8:

LANGSUNG = """
    SELECT kota, sum(nilai) AS gmv FROM pesanan GROUP BY kota
"""

DUA_TAHAP = """
    WITH parsial AS (
        SELECT kota, pesanan_id % 8 AS garam, sum(nilai) AS gmv
        FROM pesanan
        GROUP BY kota, garam
    )
    SELECT kota, sum(gmv) AS gmv FROM parsial GROUP BY kota
"""

Output yang Dihasilkan (5 run per varian, run tercepat dilaporkan):

GROUP BY langsung        tercepat 0.053 detik (run: ['0.058', '0.063', '0.080', '0.078', '0.053']) GMV Jakarta Rp71,880,000,000
GROUP BY dua tahap       tercepat 0.093 detik (run: ['0.093', '0.094', '0.100', '0.094', '0.095']) GMV Jakarta Rp71,880,000,000
hasil keduanya sama: True (20 kota, 4,000,000 baris)

Jawabannya jujur: tidak. Dua tahap justru hampir dua kali lebih lambat (0,093 versus 0,053 detik) di mesin uji. Alasannya struktural: shuffle Spark memindahkan data antar partition sehingga kunci panas menumpuk di satu task, tetapi hash aggregation DuckDB memproses tabel hash secara paralel lintas thread internal — kunci Jakarta tidak mengunci satu thread tersendiri. Salting hanya menambah satu putaran agregasi tanpa penyakit yang disembuhkan. Ringkasan keputusannya: hitung rasio puncak/rata-rata dari diagnosa; salting layak bila rasio tinggi dan engine benar-benar mengisolasi kunci panas ke satu unit eksekusi (Spark shuffle, multiprocessing); engine dengan agregasi paralel internal lebih baik dibiarkan bekerja sendiri. Fungsi hash crc32 terdokumentasi di dokumentasi zlib , dan perilaku agregasi DuckDB di dokumentasi DuckDB .

Compute kini seimbang di seluruh worker dan query. Sisa biaya besar pipeline jangka panjang justru berkembang pelan-pelan: tagihan penyimpanan yang naik setiap bulan bersama data baru — pokok bahasan chapter Mengoptimalkan Biaya Storage.