Concurrent Processing

Pipeline harian sering mengunduh enam file sumber, memanggil belasan endpoint, lalu memvalidasi hasilnya. Bila semua tugas dijalankan satu per satu, sebagian besar waktu habis hanya untuk menunggu. Python menyediakan dua jalur konkurensi — utas untuk beban I/O dan proses untuk beban CPU — dan memilih jalur yang salah membuat program nyaris tidak lebih cepat. Kita mulai dari penyebab utamanya: GIL.

GIL: Satu Kunci untuk Satu Interpreter #

CPython — implementasi Python paling umum — menjaga satu kunci bernama Global Interpreter Lock (GIL): hanya satu utas boleh mengeksekusi bytecode Python pada satu waktu dalam satu proses. Ada dua konsekuensi bagi data engineering.

Pertama, beban I/O tetap diuntungkan utas: saat menunggu respons jaringan atau isi file, utas melepas GIL sehingga utas lain bisa maju. Kedua, beban CPU murni tidak memperoleh paralelisme dari utas: kalkulasi for Python membutuhkan interpreter setiap saat, sehingga utas hanya bergantian, bukan berbarengan.

ThreadPoolExecutor untuk Tugas I/O #

Contoh pertama menyimulasikan pengunduhan enam file penjualan bulanan — pola umum pada tahap ingestion. Setiap unduhan menunggu 0,3 detik.

import time
import zlib
from concurrent.futures import ThreadPoolExecutor

BERKAS = [
    "penjualan-2024-01.csv",
    "penjualan-2024-02.csv",
    "penjualan-2024-03.csv",
    "penjualan-2024-04.csv",
    "penjualan-2024-05.csv",
    "penjualan-2024-06.csv",
]


def unduh(nama_berkas):
    # simulasi unduh: di dunia nyata ini requests.get(...)
    time.sleep(0.3)
    ukuran = 150 + zlib.crc32(nama_berkas.encode()) % 100
    return f"{nama_berkas} ({ukuran} KB)"


# 1. Berurutan
mulai = time.perf_counter()
hasil_sekuensial = [unduh(b) for b in BERKAS]
durasi_sekuensial = time.perf_counter() - mulai
print(f"sekuensial : {durasi_sekuensial:.2f} detik")

# 2. ThreadPoolExecutor
mulai = time.perf_counter()
with ThreadPoolExecutor(max_workers=6) as pool:
    hasil_pool = list(pool.map(unduh, BERKAS))
durasi_pool = time.perf_counter() - mulai
print(f"pool thread: {durasi_pool:.2f} detik")

for baris in hasil_pool:
    print("terunduh:", baris)

print("percepatan :", round(durasi_sekuensial / durasi_pool, 1), "x")

Output yang Dihasilkan:

sekuensial : 1.80 detik
pool thread: 0.30 detik
terunduh: penjualan-2024-01.csv (239 KB)
terunduh: penjualan-2024-02.csv (175 KB)
terunduh: penjualan-2024-03.csv (151 KB)
terunduh: penjualan-2024-04.csv (215 KB)
terunduh: penjualan-2024-05.csv (227 KB)
terunduh: penjualan-2024-06.csv (219 KB)
percepatan : 6.0 x

Mari kita telaah baris-baris kode di atas:

1. Baris 15-19 (unduh): time.sleep(0.3) mensimulasikan latensi jaringan; selama menunggu, utas melepas GIL. Ukuran berkas dihitung dari zlib.crc32 agar output uji dapat diulang.
2. Baris 23-26: jalur sekuensial memanggil unduh enam kali berurutan, sehingga durasinya kira-kira 6 × 0,3 = 1,8 detik.
3. Baris 29-33 (ThreadPoolExecutor): enam utas dibuat lewat max_workers=6, lalu pool.map membagi enam berkas ke seluruh utas. Semua unduhan menunggu bersamaan; durasi turun menjadi 0,30 detik — mendekati durasi satu unduhan.
4. Baris 35-38: hasil pool.map keluar dalam urutan argumen, bukan urutan selesai; kecepatan akhirnya 6,0 kali lipat.

ProcessPoolExecutor untuk Tugas CPU #

Contoh kedua memakai beban murni CPU: fungsi checksum menghitung akumulasi besar tanpa I/O. Pool proses menjalankan pekerjaan di beberapa proses OS, masing-masing dengan interpreter dan GIL sendiri.

import time
from concurrent.futures import ProcessPoolExecutor


def checksum(n):
    """Tugas berat murni CPU: akumulasi nilai besar tanpa I/O."""
    total = 0
    for i in range(n):
        total += (i * i) % 1_000_003
    return total


def tugas(n):
    # tiap pekerjaan ~0.4 detik di satu core
    return checksum(2_000_000)


DAFTAR_TUGAS = [2_000_000] * 4

if __name__ == "__main__":
    mulai = time.perf_counter()
    hasil_seq = [tugas(n) for n in DAFTAR_TUGAS]
    durasi_seq = time.perf_counter() - mulai
    print(f"sekuensial   : {durasi_seq:.2f} detik")

    mulai = time.perf_counter()
    with ProcessPoolExecutor(max_workers=2) as pool:
        hasil_par = list(pool.map(tugas, DAFTAR_TUGAS))
    durasi_par = time.perf_counter() - mulai
    print(f"pool proses  : {durasi_par:.2f} detik")

    print("hasil identik:", hasil_seq == hasil_par)
    print("percepatan   :", round(durasi_seq / durasi_par, 1), "x")
    print("checksum tugas pertama:", hasil_par[0])

Output yang Dihasilkan:

sekuensial   : 0.80 detik
pool proses  : 0.50 detik
hasil identik: True
percepatan   : 1.6 x
checksum tugas pertama: 999794999285

Output di atas dijalankan pada mesin uji dua core, sehingga percepatannya 1,6 kali — bukan empat kali seperti jumlah tugas: empat tugas antre di dua proses pekerja, ditambah biaya menyalin data antar proses. Nilai hasil identik: True menegaskan hasil paralel sama persis dengan sekuensial; angka durasi berubah sedikit antar eksekusi.

Mari kita telaah baris-baris kode di atas:

1. Baris 5-10 (checksum): perulangan murni CPU tanpa operasi menunggu — kebalikan fungsi unduh; jenis beban yang membutuhkan proses, bukan utas.
2. Baris 13-15 (tugas): pembungkus antrean. Empat entri DAFTAR_TUGAS pada baris 18 berisi parameter sama; pada pipeline nyata antrean ini bisa berisi berkas berbeda.
3. Baris 20 (if __name__ == "__main__":): guard wajib pool proses: proses anak mengimpor ulang modul saat dibuat; tanpa guard, pembuatan pool ikut dijalankan ulang dan berakhir galat rekursif.
4. Baris 27 (ProcessPoolExecutor(max_workers=2)): jumlah pekerja disamakan dengan jumlah core mesin uji. Objek yang dikirim antar proses harus picklable — angka dan string aman.

Kapan Memakai Apa #

Panduan memilih model konkurensi untuk pipeline sehari-hari:

Baris terakhir tabel sering terlewat: konkurensi bukan tujuan, melainkan alat. Ukur dulu dengan time.perf_counter(); bila versi sekuensial sudah cukup cepat, versi sederhana lebih mudah dirawat.

Rujukan resmi: dokumentasi concurrent.futures dan multiprocessing . Ketika jumlah koneksi I/O naik ke ratusan atau ribuan, utas mulai terasa mahal — chapter berikutnya, Asynchronous Programming, mengganti pendekatannya dengan asyncio.