Parallel Processing

Agregasi harian 12 ribu baris pesanan selesai dalam hitungan detik, tetapi ketika data tumbuh menjadi 1,2 miliar baris, satu mesin butuh enam jam — dan jendela pemrosesan hanya satu jam. Parallel processing memotong pekerjaan besar menjadi potongan-potongan yang dikerjakan serentak oleh banyak worker. Hasilnya bukan komputer yang lebih cepat, melainkan banyak komputer biasa yang bekerja bersamaan. Bab ini membahas cara membagi kerja, mengukur percepatan secara nyata, dan batas teoretis yang tidak bisa ditipu oleh penambahan worker.

Data Parallelism vs Task Parallelism #

Ada dua cara memotong pekerjaan, dan keduanya menjawab pertanyaan berbeda. Data parallelism memotong datanya: operasi yang sama dijalankan pada partisi-partisi berbeda secara bersamaan — inilah yang dipakai Spark saat membaca ratusan partisi Parquet. Task parallelism memotong pekerjaannya: task-task berbeda berjalan bersamaan selama tidak saling menunggu — inilah yang dilakukan orchestrator terhadap DAG, seperti dibahas pada Chapter Dependency dan Task.

Kedua pola sering dipakai berlapis: satu task agregasi dijalankan orchestrator, dan task tersebut memparalelkan pekerjaannya secara internal ke puluhan partisi. Memilih pola bergantung pada sumber kemacetan: bila yang menumpuk adalah datanya, data parallelism yang membantu; bila yang menunggu adalah tahapan pipeline, task parallelism yang mempercepat.

Benchmark Nyata: Serial vs Paralel #

Teori terasa abstrak, jadi kita ukur langsung di mesin ini (2 CPU). Beban kerjanya checksum sederhana atas 8 juta indeks — murni komputasi, tanpa akses disk, sehingga perbedaan murni efek paralelisasi. Versi serial memproses seluruh rentang dalam satu proses; versi paralel membelah rentang menjadi dua dan menyerahkan tiap belahan ke satu worker multiprocessing.Pool.

"""Benchmark nyata: checksum serial vs paralel pada mesin 2 CPU."""
import time
from multiprocessing import Pool

N = 8_000_000  # total iterasi
MOD = 1_000_003  # bilangan prima, agar hasil tetap kecil


def checksum(rentang):
    """Hitung checksum kuadrat untuk satu rentang indeks [awal, akhir)."""
    awal, akhir = rentang
    total = 0
    for i in range(awal, akhir):
        total = (total + i * i) % MOD
    return total


def ukur(label, fungsi):
    mulai = time.perf_counter()
    hasil = fungsi()
    durasi = time.perf_counter() - mulai
    print(f"{label:24s} durasi={durasi:6.3f} detik  checksum={hasil}")
    return durasi


def serial():
    return checksum((0, N))


def paralel(jumlah_worker):
    batas = [(i * N // jumlah_worker, (i + 1) * N // jumlah_worker)
             for i in range(jumlah_worker)]
    with Pool(jumlah_worker) as pool:
        hasil = sum(pool.map(checksum, batas)) % MOD
    return hasil


if __name__ == "__main__":
    t_serial = ukur("serial (1 proses)", serial)
    t_paralel = ukur("paralel (2 proses)", lambda: paralel(2))
    speedup = t_serial / t_paralel
    efisiensi = speedup / 2 * 100
    print(f"speedup = {t_serial:.3f} / {t_paralel:.3f} = {speedup:.2f}x")
    print(f"efisiensi pada 2 CPU = {efisiensi:.0f}%")

Output yang Dihasilkan:

serial (1 proses)        durasi= 0.855 detik  checksum=995103
paralel (2 proses)       durasi= 0.476 detik  checksum=995103
speedup = 0.855 / 0.476 = 1.79x
efisiensi pada 2 CPU = 90%

Angka durasi bervariasi antar eksekusi dan bergantung pada beban mesin: pengukuran di atas diambil saat mesin tidak sibuk, dan saat CPU dipakai proses lain, speedup yang terukur bisa turun ke sekitar 1,4x karena kedua worker berbagi CPU. Nilai checksum justru stabil di semua eksekusi. Mari kita telaah baris-baris kode di atas:

1. Baris 5-6 (N dan MOD): beban kerja 8 juta iterasi; hasil dijaga tetap kecil dengan operasi modulo agar checksum mudah dibandingkan.
2. Baris 9-15 (checksum): satu unit kerja murni komputasi untuk satu rentang indeks — tanpa disk dan tanpa jaringan.
3. Baris 30-35 (paralel): rentang 0 sampai N dibelah rata menjadi daftar batas, lalu pool.map menyerahkan tiap belahan ke worker; hasilnya dijumlahkan dan dimodulo ulang.
4. Baris 38-40 (if __name__): gerbang ini wajib di macOS dan Windows karena tiap worker adalah proses Python baru yang mengimpor ulang modul — tanpa gerbang, tiap worker membuat worker baru lagi tanpa henti.
5. Baris 41-44: speedup adalah perbandingan durasi serial terhadap paralel; efisiensi membagi speedup dengan jumlah CPU.

Dua checksum yang identik (995103) membuktikan pembelahan rentang tidak mengubah hasil. Speedup 1,79x pada 2 CPU berarti efisiensi 90% — sisanya habis sebagai biaya pembuatan proses dan pengiriman tugas.

Hukum Amdahl: Plafon Percepatan #

Berapa cepat bila worker ditambah terus? Hukum Amdahl menjawab: percepatan maksimum dibatasi oleh porsi kerja yang tetap serial. Bila fraksi p dari pekerjaan bisa diparalelkan dan sisanya (1 - p) serial, speedup dengan n worker adalah 1 / ((1 - p) + p / n).

"""Hitung speedup menurut Hukum Amdahl: p = fraksi kerja yang bisa diparalelkan."""


def speedup(p, n):
    """Speedup teoretis dengan n worker bila fraksi p bebas diparalelkan."""
    return 1 / ((1 - p) + p / n)


print(f"{'fraksi p':>9} {'n=2':>6} {'n=4':>6} {'n=8':>6} {'n=16':>6} {'plafon':>7}")
for p in (0.50, 0.90, 0.95, 0.99):
    sel = [f"{speedup(p, n):6.2f}" for n in (2, 4, 8, 16)]
    plafon = 1 / (1 - p)
    print(f"{p:9.2f} {' '.join(sel)} {plafon:7.1f}")

Output yang Dihasilkan:

 fraksi p    n=2    n=4    n=8   n=16  plafon
     0.50   1.33   1.60   1.78   1.88     2.0
     0.90   1.82   3.08   4.71   6.40    10.0
     0.95   1.90   3.48   5.93   9.14    20.0
     0.99   1.98   3.88   7.48  13.91   100.0

Kolom plafon adalah pesan utamanya: bila hanya 95% pekerjaan bisa diparalelkan, tidak ada gunanya menambah worker ke ratusan — percepatan mentok di 20x. Benchmark kita cocok dengan teori: pada p = 0.90 dan n = 2, Amdahl meramal 1,82x sementara hasil ukur 1,79x; selisih tipis itu adalah biaya komunikasi antar proses. Detail modul multiprocessing bisa dibaca di dokumentasi Python .

Membagi kerja dan mengukurnya kini tuntas. Namun ketika banyak worker menulis ke banyak database secara bersamaan, muncul pertanyaan baru: bagaimana memastikan serangkaian penulisan lintas node terjadi seluruhnya atau tidak sama sekali. Chapter Distributed Transactions membahas two-phase commit dan pola saga sebagai jawabannya.