Parallelism

Pipeline agregasi harian TokoKita membaca delapan partisi csv.gz berisi satu juta baris pesanan per partisi, lalu menghitung GMV per kota. Dibaca satu per satu, delapan unit kerja itu menghabiskan 3,45 detik sementara sebagian besar kapasitas prosesor menganggur. Chapter Concurrent Processing telah memperkenalkan ThreadPoolExecutor dan ProcessPoolExecutor, dan chapter Parallel Processing menambahkan hukum Amdahl sebagai plafon teoretis percepatan. Bab ini menerapkan keduanya pada pipeline sungguhan di mesin kecil 2 CPU — termasuk hasil ukur yang jujur ketika jumlah worker melebihi jumlah core.

Satu Unit Kerja yang Bisa Dibagi #

Paralelisme dimulai dari satu keputusan desain: pekerjaan harus bisa dipotong menjadi unit yang independen. Delapan partisi file memenuhi syarat itu — setiap partisi dibaca, difilter, dan diagregasi tanpa bergantung pada partisi lain, dan hasil parsial per kota tinggal dijumlahkan di proses induk. Skrip berikut mengukur tiga cara menjalankan delapan unit kerja yang sama:

BERKAS = sorted(glob.glob("data/ch7/*.csv.gz"))


def transform(path):
    """Satu unit kerja: baca satu partisi csv.gz, agregasi GMV per kota."""
    df = pd.read_csv(path, compression="gzip")
    df = df[df["qty"] > 0]
    return df.assign(total=df["qty"] * df["harga"]).groupby("kota")["total"].sum()


def gabung(hasil_per_file):
    total = {}
    for hasil in hasil_per_file:
        for kota, gmv in hasil.items():
            total[kota] = total.get(kota, 0) + gmv
    return total


def ukur(label, jalan, ulang=3):
    tercepat = None
    for _ in range(ulang):
        t0 = time.perf_counter()
        hasil = jalan()
        durasi = time.perf_counter() - t0
        if tercepat is None or durasi < tercepat:
            tercepat = durasi
    print(f"{label:28s} {tercepat:6.2f} detik")
    return tercepat, hasil


if __name__ == "__main__":
    t_seq, hasil_seq = ukur("sekuensial (1 proses)",
                            lambda: gabung([transform(f) for f in BERKAS]))

    for n in (2, 4):
        def thread(n=n):
            with ThreadPoolExecutor(max_workers=n) as pool:
                return gabung(list(pool.map(transform, BERKAS)))
        ukur(f"ThreadPool ({n} worker)", thread)

    for n in (2, 4, 8):
        def proses(n=n):
            with ProcessPoolExecutor(max_workers=n) as pool:
                return gabung(list(pool.map(transform, BERKAS)))
        t, hasil = ukur(f"ProcessPool ({n} worker)", proses)
        if n == 2:
            print(f"  -> ProcessPool(2) {t_seq / t:.1f}x lebih cepat "
                  f"dari sekuensial; hasil identik: {hasil == hasil_seq}")

Docstring, import, dan pembuatan delapan partisi dipangkas dari tampilan. Mesin uji memiliki 2 CPU; tiap konfigurasi diulang tiga kali dan dilaporkan run tercepat, karena derau eksternal hanya bisa menambah waktu, tidak mengurangi.

Output yang Dihasilkan:

sekuensial (1 proses)          3.45 detik
ThreadPool (2 worker)          2.82 detik
ThreadPool (4 worker)          2.38 detik
ProcessPool (2 worker)         2.57 detik
  -> ProcessPool(2) 1.3x lebih cepat dari sekuensial; hasil identik: True
ProcessPool (4 worker)         2.58 detik
ProcessPool (8 worker)         3.08 detik

Mari kita telaah bagian-bagian penting kode di atas:

1. Baris 15-19 (transform): satu unit kerja membaca satu partisi, membuang baris bermasalah, lalu mengembalikan GMV per kota; seluruh parsing berlangsung di kode C milik pandas.
2. Baris 22-27 (gabung): proses induk menjumlahkan hasil parsial; karena penjumlahan komutatif, urutan penggabungan bebas dipilih.
3. Baris 30-39 (ukur): tiga pengulangan per konfigurasi dengan time.perf_counter(), dan hanya nilai minimum yang dilaporkan.
4. Baris 43-44 (baseline sekuensial): list comprehension memaksa delapan unit kerja berjalan bergiliran dalam satu proses.
5. Baris 51 dan 57 (parameter n=n): nilai n ditangkap saat definisi fungsi, menghindari jebakan late-binding closure ketika fungsi baru dipanggil setelah loop berakhir.

Benang dan Proses di Mesin 2 CPU #

Hasil pertama yang mengejutkan: ThreadPoolExecutor justru unggul. Empat utas menyelesaikan pekerjaan dalam 2,38 detik, lebih cepat daripada dua proses terpisah (2,57 detik) — padahal tabel panduan chapter Concurrent Processing mengarahkan beban parsing ke ProcessPoolExecutor. Penjelasannya ada di dalam pd.read_csv: parsing berjalan di kode C yang melepaskan GIL selama bekerja, dan pembukaan file gzip menyisakan waktu tunggu I/O. Utas-utas saling mengisi waktu tunggu itu, sementara pool proses membayar biaya start-up interpreter dan serialisasi hasil pada mesin yang sama-sama hanya punya dua core.

Hasil kedua adalah peringatan klasik: menambah worker di atas jumlah core memperburuk keadaan. ProcessPool(8) melambat menjadi 3,08 detik karena delapan proses berebut dua core — biaya penjadwalan dibayar tanpa kapasitas baru. Itulah oversubscription. Dibandingkan ramalan Amdahl pada chapter Parallel Processing (speedup 1,82x untuk p = 0,90 dan n = 2), pengukuran kita hanya mencapai 1,3x. Selisihnya bukan kesalahan teori: Amdahl adalah plafon tanpa biaya overhead, sedangkan pipeline nyata membayar start-up proses, serialisasi, dan antrean disk yang sama.

DuckDB Mengatur Sendiri Paralelnya #

Pipeline yang berbentuk SQL tidak perlu dipotong manual. DuckDB memparalelkan satu query lintas thread secara internal, sehingga pekerjaan kita tinggal menyetel threads dan menjalankan query yang sama pada delapan partisi yang sama:

"""Benchmark duckdb threads ch7: query yang sama, threads 1/2/4/8."""
import time

import duckdb

con = duckdb.connect()
con.execute("SET enable_progress_bar=false")

QUERY = """
    SELECT kota, sum(qty * harga) AS gmv
    FROM read_csv('data/ch7/*.csv.gz')
    GROUP BY kota ORDER BY kota
"""

if __name__ == "__main__":
    for n in (1, 2, 4, 8):
        con.execute(f"SET threads={n}")
        runs = []
        for _ in range(3):
            t0 = time.perf_counter()
            hasil = con.execute(QUERY).fetchall()
            runs.append(time.perf_counter() - t0)
        print(f"threads={n}: tercepat {min(runs):.2f} detik "
              f"(3 run: {[f'{r:.2f}' for r in runs]})")
    print("GMV Jakarta:", f"{dict(hasil)['Jakarta']:,.0f}")

Output yang Dihasilkan:

threads=1: tercepat 2.95 detik (3 run: ['3.09', '2.97', '2.95'])
threads=2: tercepat 2.03 detik (3 run: ['2.13', '2.10', '2.03'])
threads=4: tercepat 2.18 detik (3 run: ['2.18', '2.20', '2.71'])
threads=8: tercepat 2.56 detik (3 run: ['2.58', '2.56', '2.60'])
GMV Jakarta: 2,072,700,463,000

Mari kita telaah baris-baris kode di atas:

1. Baris 9-13 (QUERY): read_csv dengan pola glob memuat delapan partisi sebagai satu tabel; agregasinya ditulis satu kalimat.
2. Baris 16-24 (loop threads): tiap konfigurasi dijalankan tiga kali, dan daftar run asli ikut dicetak supaya sebarannya terlihat — bukan hanya angka terbaiknya.
3. Baris 25 (GMV Jakarta): pemeriksaan kewarasan bahwa query memproses seluruh delapan partisi, bukan sebagian.

Kurva yang sama muncul: threads=2 tercepat dengan speedup 1,45x dari threads=1, lalu threads=4 dan threads=8 melambat karena pembagian kerja dan sinkronisasi antar thread mulai dibayar. Pengaturan thread dan pembacaan CSV DuckDB selengkapnya ada di dokumentasi DuckDB .

Panduan Pemilihan Worker #

Satu jebakan masih tertinggal dari keberhasilan membagi delapan partisi: bila satu partisi jauh lebih berat dari yang lain, worker termahal menahan seluruh pipeline sendirian — masalah data skew yang kita bedah pada chapter Mengatasi Data Skew. Sebelum ke sana, ada cara yang lebih murah daripada menghitung lebih cepat: tidak menghitung sama sekali pekerjaan yang hasilnya sudah tersedia — pokok bahasan chapter Caching. Rujukan concurrent.futures resmi ada di dokumentasi Python .