Bab 67 — Sistem Terdistribusi: Banyak Worker, Satu Pekerjaan Besar
Seri A5 — Python Tingkat Ahli
Semua skrip bab ini ada di folder kode/bab67/. Di versi web, Anda juga bisa mengunduh semuanya sekaligus (zip).
Di Bab 66 Anda mempercepat satu komputer. Sekarang bayangkan sebuah kebun besar yang baru saja diterbangi drone: ribuan foto menumpuk, ditambah citra satelit tiap minggu. Satu orang secepat apa pun tidak akan selesai tepat waktu. Solusinya seperti dapur umum saat panen raya: ada papan pesanan (antrean), ada banyak juru masak (worker), dan juru masak mana pun yang kosong mengambil pesanan berikutnya. Kalau satu juru masak sakit, pesanannya diambil yang lain. Itulah sistem terdistribusi: pekerjaan dibagi ke banyak pelaksana yang berkoordinasi lewat perantara.
Aturan bab ini. Untuk menjaga build buku tetap cepat, bahan latihan kita hanya 48 berkas tiruan (36 foto drone dan 12 citra satelit, dibuat oleh skrip, tanpa unduhan apa pun), bukan ribuan; ide dan kodenya sama untuk ribuan. Semua skrip yang memakai antrean dijalankan sungguhan dengan Redis 5.0.14 sementara (arahan # @redis), dengan worker yang dinyalakan dan dimatikan oleh skrip itu sendiri. Waktu proses tidak dicetak; yang dicetak hanya hasil yang stabil, dan angka di komputer Anda berbeda.
Bahan latihan: foto drone dan citra satelit tiruan
01_citra_tiruan.py
# Bab 67 - Bahan latihan: foto drone dan citra satelit TIRUAN, dan analisis per berkas
import shutil
from pathlib import Path
from citra_tiruan import analisis, buat_citra
folder = Path(__file__).parent / "_tmp" / "citra"
shutil.rmtree(folder, ignore_errors=True)
berkas = buat_citra(folder, jumlah_drone=36, jumlah_satelit=12)
print("jumlah berkas :", len(berkas))
print("foto drone :", sum(b.endswith(".png") for b in berkas), "(PNG, 256 x 256, tiga pita warna)")
print("citra satelit :", sum(b.endswith(".npz") for b in berkas), "(NPZ, 256 x 256, pita merah dan NIR)")
print("\ncontoh hasil analisis (indeks vegetasi per berkas):")
for jalur in (berkas[0], berkas[1], berkas[2], berkas[36], berkas[37], berkas[38]):
h = analisis(jalur)
print(f" {h['berkas']:<18} {h['jenis']:<8} petak {h['petak']} rata={h['rata']:.4f} hijau={h['persen_hijau']:>5.1f}%")
# Hasilnya bersifat tetap (benih acak tetap): dua kali dihitung, hasilnya sama
print("\ndihitung ulang, hasil sama:", analisis(berkas[0]) == analisis(berkas[0]))jumlah berkas : 48
foto drone : 36 (PNG, 256 x 256, tiga pita warna)
citra satelit : 12 (NPZ, 256 x 256, pita merah dan NIR)
contoh hasil analisis (indeks vegetasi per berkas):
drone_A_000.png drone petak A rata=0.4992 hijau= 49.6%
drone_B_001.png drone petak B rata=0.3550 hijau= 20.0%
drone_C_002.png drone petak C rata=0.1922 hijau= 0.0%
satelit_A_000.npz satelit petak A rata=0.4547 hijau= 83.6%
satelit_B_001.npz satelit petak B rata=0.3172 hijau= 54.3%
satelit_C_002.npz satelit petak C rata=0.2433 hijau= 28.9%
dihitung ulang, hasil sama: TrueModul citra_tiruan.py (ikut di folder dan zip) membuat 36 foto drone (PNG 256 x 256, tiga pita) dan 12 citra satelit (dua pita: merah dan NIR). Petak A dibuat paling rimbun, C paling jarang. Fungsi analisis menghitung indeks vegetasi (Excess Green untuk foto drone, NDVI untuk satelit; Bab 59), lalu menghaluskannya berulang kali supaya pekerjaannya cukup berat (sekitar 0,1 detik per berkas) seperti tahap penyaringan sungguhan. Hasilnya dict kecil dan tetap (benih acak tetap). Perhatikan urutan: drone A lebih hijau dari B, B dari C, dan seterusnya.
Antrean paling sederhana: sebuah daftar di Redis
Redis adalah basis data dalam memori yang sangat cepat dan punya tipe list. Itu cukup untuk antrean: produser LPUSH tugas di satu ujung, worker BRPOP (ambil sambil menunggu) di ujung lain.
02_antrean_redis_dasar.py
# Bab 67 - Antrean tugas paling sederhana: daftar (list) di Redis
import json
import multiprocessing as mp
import os
import redis
URL = os.environ["REDIS_URL"]
ANTREAN, SELESAI, HASIL = "bab67:02:antrean", "bab67:02:selesai", "bab67:02:hasil"
def klien():
return redis.Redis.from_url(URL, protocol=2, decode_responses=True) # protocol=2: Redis 5
def pekerja(nama):
r = klien()
while True:
_, isi = r.brpop(ANTREAN) # menunggu sampai ada tugas
tugas = json.loads(isi)
if tugas == "BERHENTI":
return
r.hset(HASIL, tugas["id"], sum(i * i for i in range(tugas["n"])))
r.rpush(SELESAI, nama) # catat siapa yang mengerjakan
if __name__ == "__main__":
r = klien()
r.delete(ANTREAN, SELESAI, HASIL)
for i in range(20): # produser: memasukkan 20 tugas
r.lpush(ANTREAN, json.dumps({"id": i, "n": 1000 + i}))
print("tugas menunggu di antrean:", r.llen(ANTREAN))
proses = [mp.Process(target=pekerja, args=(f"w{k}",)) for k in range(3)]
for p in proses:
p.start()
while r.llen(SELESAI) < 20: # tunggu semua selesai
pass
for _ in proses:
r.lpush(ANTREAN, json.dumps("BERHENTI")) # satu tanda berhenti per pekerja
for p in proses:
p.join()
hasil = {int(k): int(v) for k, v in r.hgetall(HASIL).items()}
benar = all(hasil[i] == sum(j * j for j in range(1000 + i)) for i in range(20))
print("tugas selesai :", len(hasil), "dari 20; semua hasil benar:", benar)
print("antrean sekarang :", r.llen(ANTREAN))
# Masalah: tugas yang sudah diambil lalu pekerjanya mati akan HILANG.
# Pola andal: pindahkan tugas ke daftar "sedang diproses" lebih dulu (BRPOPLPUSH).
PROSES = "bab67:02:proses"
r.delete(ANTREAN, PROSES)
r.lpush(ANTREAN, json.dumps({"id": 99, "n": 10}))
diambil = r.brpoplpush(ANTREAN, PROSES) # diambil, tetapi tercatat sedang diproses
print("pekerja mati sebelum selesai; antrean:", r.llen(ANTREAN), "| sedang diproses:", r.llen(PROSES))
while r.rpoplpush(PROSES, ANTREAN): # pemulih: kembalikan tugas yang menggantung
pass
print("setelah dipulihkan; antrean:", r.llen(ANTREAN), "| sedang diproses:", r.llen(PROSES))
r.delete(ANTREAN, PROSES, SELESAI, HASIL)tugas menunggu di antrean: 20
tugas selesai : 20 dari 20; semua hasil benar: True
antrean sekarang : 0
pekerja mati sebelum selesai; antrean: 0 | sedang diproses: 1
setelah dipulihkan; antrean: 1 | sedang diproses: 020 tugas dimasukkan, tiga proses worker mengambilnya, dan semua 20 hasil benar tanpa ada yang dikerjakan dua kali (BRPOP memberikan tiap tugas ke tepat satu worker). Setelah itu, tanda "BERHENTI" dikirim satu per worker, cara umum mematikan worker dengan sopan.
Bagian akhir menunjukkan mengapa antrean siap-pakai itu perlu: bila worker mengambil tugas lalu mati, tugas itu hilang karena sudah keluar dari antrean. Pola yang andal memindahkan tugas ke daftar “sedang diproses” (BRPOPLPUSH); bila worker mati, tugas yang menggantung dikembalikan ke antrean oleh “pemulih”. Celery dan RQ menerapkan gagasan serupa untuk Anda.
Catatan perangkat: klien redis-py versi baru berbicara protokol RESP3 dan mengirim perintah HELLO, yang tidak dikenal Redis 5 di komputer penulis. Karena itu skrip membuat klien dengan protocol=2. Di Redis versi baru Anda tidak perlu protocol=2.
Celery: produser, broker, worker, dan backend hasil
Celery adalah kerangka antrean tugas paling terkenal di Python. Empat peran:
- Produser: kode yang memanggil
tugas.delay(...); ia langsung kembali dan tidak menunggu pekerjaan. - Broker: perantara yang menyimpan pesan tugas (di sini Redis).
- Worker: proses yang mengambil pesan dan menjalankan fungsinya.
- Backend hasil: tempat hasil disimpan agar produser dapat mengambilnya (di sini juga Redis).
Dua catatan jujur dari dokumentasi dan percobaan kami. (1) FAQ resmi Celery menyatakan Windows tidak didukung sejak Celery 4 karena kekurangan sumber daya, tetapi “mungkin masih jalan” (docs.celeryq.dev, dibaca 2026-10-10). Di komputer penulis, kombinasi pool solo terbukti berjalan; pool prefork (bawaan) kami tidak coba. Setiap worker solo adalah satu proses Python dengan satu tugas pada satu waktu, jadi banyak worker = banyak proses, bukan banyak thread. (2) Agar Celery bicara dengan Redis 5, modul bantu redis5_tambalan.py menanamkan protocol=2 pada kolam sambungan redis-py (Celery membuat sambungannya sendiri, jadi argumen protocol=2 tidak bisa kita berikan langsung). Sebelum tambalan itu, worker kami menolak tersambung dengan pesan unknown command HELLO.
03_celery_worker.py
# Bab 67 - Celery: menyalakan worker, mengirim tugas, mengumpulkan hasil, mematikan worker
import os
import shutil
from pathlib import Path
os.environ["ANTREAN"] = "bab67_03" # antrean khusus skrip ini (dibaca oleh tugas_celery)
from citra_tiruan import analisis, buat_citra
from peluncur import matikan, mulai_pekerja, tunggu_siap
from tugas_celery import analisis_tugas, app
folder = Path(__file__).parent / "_tmp" / "citra03"
shutil.rmtree(folder, ignore_errors=True)
berkas = buat_citra(folder, jumlah_drone=12, jumlah_satelit=6) # 18 berkas
proses = mulai_pekerja(jumlah=3, antrean="bab67_03") # tiga worker = tiga proses Python
nama = []
try:
nama = tunggu_siap(app, 3)
print("worker yang menjawab ping :", len(nama))
# Produser: kirim tugas. delay() langsung kembali; pekerjaan terjadi di worker.
pesanan = [analisis_tugas.delay(b) for b in berkas]
print("tugas dikirim :", len(pesanan))
hasil = [p.get(timeout=120) for p in pesanan] # tunggu dan ambil hasil
print("tugas berhasil :", sum(p.successful() for p in pesanan))
print("hasil worker = hasil lokal:", hasil[0] == analisis(berkas[0]))
print("contoh hasil :", hasil[0])
finally:
matikan(proses, app, nama) # worker dimatikan oleh skrip ini
print("semua worker berhenti :", all(p.poll() is not None for p in proses))worker yang menjawab ping : 3
tugas dikirim : 18
tugas berhasil : 18
hasil worker = hasil lokal: True
contoh hasil : {'berkas': 'drone_A_000.png', 'jenis': 'drone', 'petak': 'A', 'rata': 0.4992, 'persen_hijau': 49.6}
semua worker berhenti : TrueUrutannya: skrip menyalakan tiga proses worker lewat subprocess dan menunggu mereka menjawab ping; produser mengirim 18 tugas dengan delay(); p.get() menunggu tiap hasil; hasil dari worker sama persis dengan hasil lokal; terakhir worker dimatikan dengan perintah shutdown yang ditujukan hanya ke worker kita (parameter destination; tanpanya, perintah itu menjangkau semua worker di Redis yang sama). Konfigurasi yang kami pakai di tugas_celery.py: worker_prefetch_multiplier=1 (satu tugas per worker, supaya pembagian adil) dan task_acks_late=True (tugas dianggap selesai setelah berhasil, bukan saat diambil).
Kegagalan itu pasti terjadi
Di sistem terdistribusi, yang tidak mungkin pun terjadi: jaringan putus, berkas rusak, worker mati. Perancangan yang baik menganggap kegagalan sebagai hal biasa.
04_coba_ulang_dan_gagal.py
# Bab 67 - Kegagalan itu pasti terjadi: coba-ulang otomatis dan tugas yang tetap gagal
import os
os.environ["ANTREAN"] = "bab67_04"
import redis
from peluncur import matikan, mulai_pekerja, tunggu_siap
from tugas_celery import URL, app, selalu_gagal, tidak_stabil
klien = redis.Redis.from_url(URL, protocol=2)
klien.delete("bab67:04:percobaan_a", "bab67:04:percobaan_b")
proses = mulai_pekerja(jumlah=1, antrean="bab67_04")
nama = []
try:
nama = tunggu_siap(app, 1)
# 1. Gagal dua kali karena "jaringan putus", lalu berhasil. Celery mencoba ulang sendiri
# (autoretry_for=ConnectionError, maksimum 4 kali ulang).
a = tidak_stabil.delay("bab67:04:percobaan_a", 2)
print("1. status akhir :", a.get(timeout=60)["percobaan"], "percobaan -> berhasil;", a.state)
# 2. Gagal terus-menerus: setelah batas ulang habis, tugas berstatus FAILURE
b = tidak_stabil.delay("bab67:04:percobaan_b", 99)
try:
b.get(timeout=60)
except ConnectionError as e:
print("2. gagal terus :", b.state, "-", e)
print(" percobaan tercatat:", int(klien.get("bab67:04:percobaan_b")), "(1 awal + 4 ulang)")
# 3. Kesalahan yang bukan gangguan sesaat tidak boleh diulang buta: langsung gagal
c = selalu_gagal.delay("drone_C_099.png")
try:
c.get(timeout=60)
except ValueError as e:
print("3. kesalahan data:", c.state, "-", e)
# 4. Kegagalan harus dicatat supaya bisa diperiksa orang (mirip "dead letter queue")
for hasil in (b, c):
klien.rpush("bab67:04:gagal", f"{hasil.id[:8]}:{hasil.state}")
print("4. tugas gagal tercatat untuk diperiksa:", klien.llen("bab67:04:gagal"))
finally:
matikan(proses, app, nama)
klien.delete("bab67:04:percobaan_a", "bab67:04:percobaan_b", "bab67:04:gagal")1. status akhir : 3 percobaan -> berhasil; SUCCESS
2. gagal terus : FAILURE - koneksi putus pada percobaan 5
percobaan tercatat: 5 (1 awal + 4 ulang)
3. kesalahan data: FAILURE - berkas drone_C_099.png rusak
4. tugas gagal tercatat untuk diperiksa: 2- Gangguan sesaat: tugas gagal dua kali lalu berhasil pada percobaan ketiga, dan Celery mengulangnya sendiri (
autoretry_foruntukConnectionError). - Gagal terus: setelah batas ulang habis (kami menetapkan 4), status menjadi
FAILUREsetelah total 5 percobaan. - Kesalahan data (berkas rusak) tidak ada gunanya diulang buta, jadi
ValueErrortidak masuk daftar ulang dan langsung gagal. - Catat yang gagal untuk diperiksa orang, seperti dead letter queue (“kotak surat mati”).
Dokumentasi Celery menegaskan bahwa fungsi tugas sebaiknya idempoten: bila dijalankan dua kali dengan argumen sama, tidak boleh menimbulkan efek ganda (misalnya mengirim dua kali, atau menjumlahkan dua kali ke total). Dengan acks_late, bila worker mati di tengah jalan, tugas dapat dijalankan ulang. Rancang tugas agar hasilnya ditimpa, bukan ditambah (menulis ke hasil/foto_017.json, bukan menambah baris di akhir berkas).
RQ: antrean tugas yang lebih sederhana
RQ (Redis Queue) lebih ringan: hanya Redis, tanpa konfigurasi rumit. Tugas adalah fungsi Python biasa yang dirujuk lewat namanya, mis. "rq_tugas.proses".
05_rq.py
# Bab 67 - RQ: antrean tugas yang lebih sederhana (hanya Redis), tugas gagal dan coba-ulang
import os
import shutil
import subprocess
import sys
from pathlib import Path
import redis
from rq import Queue, Retry
import redis5_tambalan # noqa: F401
from citra_tiruan import analisis, buat_citra
URL = os.environ["REDIS_URL"]
sambungan = redis.Redis.from_url(URL, protocol=2)
antrean = Queue("bab67_05", connection=sambungan)
antrean.empty()
folder = Path(__file__).parent / "_tmp" / "citra05"
shutil.rmtree(folder, ignore_errors=True)
berkas = buat_citra(folder, jumlah_drone=8, jumlah_satelit=4) # 12 berkas
# Produser: tugas ditulis sebagai teks "modul.fungsi" supaya worker yang mengimpornya
pekerjaan = [antrean.enqueue("rq_tugas.proses", b, retry=Retry(max=2, interval=0)) for b in berkas]
rusak = antrean.enqueue("rq_tugas.rusak", "drone_X.png", retry=Retry(max=2, interval=0))
print("tugas menunggu :", len(antrean), "| status tugas pertama:", pekerjaan[0].get_status().value)
# Dua worker dinyalakan (mode burst: berhenti sendiri saat antrean kosong)
folder_modul = Path(__file__).parent
worker = [subprocess.Popen([sys.executable, "rq_tugas.py", "bab67_05"], cwd=folder_modul,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(2)]
for w in worker:
w.wait(timeout=240)
print("worker berhenti sendiri :", all(w.poll() == 0 for w in worker))
for p in pekerjaan + [rusak]:
p.refresh()
berhasil = [p for p in pekerjaan if p.get_status().value == "finished"]
print("tugas berhasil :", len(berhasil), "dari", len(pekerjaan))
print("hasil sama dengan lokal :", berhasil[0].return_value() == analisis(berkas[0]))
print("tugas rusak berstatus :", rusak.get_status().value)
print(" tercatat di daftar gagal:", len(antrean.failed_job_registry), "tugas")
print(" penyebab :", rusak.latest_result().exc_string.strip().splitlines()[-1])
antrean.empty()
for p in pekerjaan + [rusak]:
p.delete()tugas menunggu : 13 | status tugas pertama: queued
worker berhenti sendiri : True
tugas berhasil : 12 dari 12
hasil sama dengan lokal : True
tugas rusak berstatus : failed
tercatat di daftar gagal: 1 tugas
penyebab : ValueError: berkas drone_X.png rusakDua worker (proses terpisah) menghabiskan 12 tugas lalu berhenti sendiri (mode burst). Tugas yang selalu gagal masuk registri gagal (failed job registry) bersama penyebabnya, sehingga bisa diperiksa dan diantrekan ulang setelah diperbaiki. Catatan Windows: menurut dokumentasi RQ, worker standar memakai fork() yang tidak ada di Windows; tersedia SimpleWorker (menjalankan tugas di proses worker itu sendiri, tanpa fork) dan, sejak RQ 2.2.0, SpawnWorker yang memakai os.spawn() (python-rq.org/docs/workers, dibaca 2026-10-10). Kami memakai SimpleWorker (versi RQ terpasang 2.12.0) dan ia berjalan dengan Redis 5. Dokumentasi menyebut SimpleWorker tidak dianjurkan untuk produksi kecuali Anda memahami batasannya (mis. tanpa detak jantung selama tugas berjalan); SpawnWorker tidak kami uji.
| Celery | RQ | |
|---|---|---|
| Broker | Redis, RabbitMQ, dan lain-lain | hanya Redis |
| Fitur | jadwal berkala, rantai tugas, banyak jenis pool, pemantauan | sederhana: antrean, coba-ulang, registri gagal |
| Kerumitan | lebih tinggi | lebih rendah |
| Windows | tidak resmi didukung; solo berjalan di sini |
tidak memakai fork: SimpleWorker berjalan di sini |
Aturan praktis: mulai dengan RQ bila kebutuhan Anda sederhana dan sudah ada Redis; pilih Celery bila perlu jadwal, rantai tugas, atau broker lain.
Pemantauan
Sistem yang tidak dipantau akan memberi tahu Anda bahwa ia rusak lewat keluhan pengguna. Tiga hal dasar dipantau: panjang antrean (tumbuh terus berarti worker kurang), status worker (hidup atau tidak), dan tugas gagal.
06_pemantauan.py
# Bab 67 - Pemantauan: panjang antrean, worker aktif, dan statistik pekerjaan
import os
import shutil
from pathlib import Path
os.environ["ANTREAN"] = "bab67_06"
import redis
from citra_tiruan import buat_citra
from peluncur import matikan, mulai_pekerja, tunggu_siap
from tugas_celery import URL, analisis_tugas, app
folder = Path(__file__).parent / "_tmp" / "citra06"
shutil.rmtree(folder, ignore_errors=True)
berkas = buat_citra(folder, jumlah_drone=12, jumlah_satelit=4) # 16 berkas
klien = redis.Redis.from_url(URL, protocol=2)
proses = mulai_pekerja(jumlah=2, antrean="bab67_06")
nama = []
try:
nama = tunggu_siap(app, 2)
pemeriksa = app.control.inspect(timeout=2)
print("tugas terdaftar di worker :", sorted(t for t in next(iter(pemeriksa.registered().values()))
if t.startswith("citra.")))
pesanan = [analisis_tugas.delay(b) for b in berkas]
antre = klien.llen("bab67_06") # antrean Celery = daftar Redis bernama sama
print("masih ada tugas menunggu :", antre > 0)
for p in pesanan:
p.get(timeout=120)
print("antrean setelah selesai :", klien.llen("bab67_06"))
statistik = pemeriksa.stats()
total = sum(s["total"].get("citra.analisis", 0) for s in statistik.values())
print("total tugas dikerjakan :", total, "(jumlah dari semua worker)")
print("kunci statistik worker :", sorted(next(iter(statistik.values())))[:6])
finally:
matikan(proses, app, nama)tugas terdaftar di worker : ['citra.analisis', 'citra.analisis_terukur', 'citra.selalu_gagal', 'citra.tidak_stabil']
masih ada tugas menunggu : True
antrean setelah selesai : 0
total tugas dikerjakan : 16 (jumlah dari semua worker)
kunci statistik worker : ['broker', 'clock', 'pid', 'pool', 'prefetch_count', 'rusage']Antrean Celery di Redis adalah daftar Redis bernama sama, jadi LLEN memberi panjangnya (cara yang juga dianjurkan dokumentasi pemantauan Celery untuk broker Redis). inspect().registered() menampilkan tugas yang dikenali worker, dan inspect().stats() memberi hitungan tugas per worker: jumlahnya 16, persis tugas yang kita kirim. Satu catatan penting dari dokumentasi: pada pool solo, sebuah tugas yang sedang berjalan memblokir perintah kendali yang menunggu, sehingga worker yang sibuk menjawab ping atau inspect dengan lambat (naikkan batas waktu balasan). Untuk tampilan web, Celery punya Flower (celery -A proj flower, port 5555; tidak kami jalankan untuk buku ini).
Desain layanan pemroses citra
Rangkaian prinsip yang membuat sistem seperti di gambar atas tahan banting:
- Kirim rujukan, bukan data besar. Pesan antrean berisi alamat berkas atau ID (
drone_A_000.png), bukan isi gambar. Gambar disimpan di penyimpanan bersama (disk jaringan atau penyimpanan objek). - Idempoten. Satu tugas boleh berjalan dua kali tanpa akibat buruk; hasil ditulis dengan menimpa.
- Hasil kecil di antrean, hasil besar di penyimpanan. Dict ringkas (seperti hasil kita) boleh lewat backend; peta atau citra hasil ditulis ke penyimpanan dan hanya alamatnya yang dikirim balik.
- Batas waktu dan coba-ulang yang wajar, serta pencatatan tugas gagal.
- Satu antrean per jenis pekerjaan: tugas ringan tidak antre di belakang tugas berat; worker GPU (Bab 66) hanya melayani antrean yang butuh GPU.
- Tekanan balik (backpressure): batasi laju produser bila antrean terlalu panjang.
- Pemantauan dan peringatan: antrean menumpuk atau banyak tugas gagal harus memicu tanda bahaya.
- Versi tugas dan skema pesan: worker lama dan baru sering hidup bersamaan saat rilis.
Skalabilitas dan biaya
Menambah worker tidak mempercepat tanpa batas. Hukum Amdahl menyatakan bila sebagian pekerjaan tidak bisa dibagi (membaca daftar berkas, menyatukan hasil), percepatan maksimum dibatasi bagian itu: dengan 2% bagian serial, batasnya 50 kali berapa pun jumlah worker. Skrip berikut menghitung dengan tarif tiruan (bukan harga penyedia mana pun).
07_biaya_dan_skala.py
# Bab 67 - Biaya dan skalabilitas: hitungan dengan tarif TIRUAN (bukan harga penyedia mana pun)
# Asumsi (semua angka contoh):
FOTO = 100_000 # foto drone + citra satelit yang harus diproses
DETIK_PER_FOTO_CPU = 2.0 # satu inti CPU memproses satu foto
DETIK_PER_FOTO_GPU = 0.1 # satu GPU memproses satu foto (data sudah di kartu)
TARIF_CPU_JAM = 3_000 # Rp per jam untuk satu mesin 4 inti (tiruan)
TARIF_GPU_JAM = 9_000 # Rp per jam untuk satu mesin dengan GPU (tiruan)
INTI_PER_MESIN = 4
BAGIAN_SERIAL = 0.02 # 2% pekerjaan tidak bisa dibagi (membaca daftar, menyatukan hasil)
def percepatan_amdahl(pekerja, bagian_serial):
"""Hukum Amdahl: percepatan maksimum bila sebagian pekerjaan tidak bisa diparalelkan."""
return 1 / (bagian_serial + (1 - bagian_serial) / pekerja)
print(f"{'worker':>7} {'percepatan':>11} {'efisiensi':>10}")
for w in (1, 2, 4, 8, 16, 32, 64):
s = percepatan_amdahl(w, BAGIAN_SERIAL)
print(f"{w:>7} {s:>10.1f}x {s / w:>9.0%}")
print("batas teoritis (tak terhingga worker):", round(1 / BAGIAN_SERIAL), "x")
def rupiah(angka):
"""Rp dengan titik sebagai pemisah ribuan, gaya Indonesia."""
return "Rp " + f"{angka:,.0f}".replace(",", ".")
print("\nSkenario untuk", f"{FOTO:,}".replace(",", "."), "foto:")
total_detik_cpu = FOTO * DETIK_PER_FOTO_CPU
for mesin in (1, 4, 16):
inti = mesin * INTI_PER_MESIN
jam = total_detik_cpu / inti / 3600
biaya = mesin * jam * TARIF_CPU_JAM
print(f" CPU, {mesin:>2} mesin ({inti:>2} inti): {jam:6.1f} jam, biaya {rupiah(biaya):>10}")
jam_gpu = FOTO * DETIK_PER_FOTO_GPU / 3600
print(f" GPU, 1 mesin : {jam_gpu:6.1f} jam, biaya {rupiah(jam_gpu * TARIF_GPU_JAM):>10}")
# Pelajaran: biaya CPU hampir tetap (jam x mesin sama), tetapi waktu tunggu turun
print("\nBiaya CPU total (jam-mesin dikali tarif) praktis sama; yang berubah adalah lamanya menunggu.") worker percepatan efisiensi
1 1.0x 100%
2 2.0x 98%
4 3.8x 94%
8 7.0x 88%
16 12.3x 77%
32 19.8x 62%
64 28.3x 44%
batas teoritis (tak terhingga worker): 50 x
Skenario untuk 100.000 foto:
CPU, 1 mesin ( 4 inti): 13.9 jam, biaya Rp 41.667
CPU, 4 mesin (16 inti): 3.5 jam, biaya Rp 41.667
CPU, 16 mesin (64 inti): 0.9 jam, biaya Rp 41.667
GPU, 1 mesin : 2.8 jam, biaya Rp 25.000
Biaya CPU total (jam-mesin dikali tarif) praktis sama; yang berubah adalah lamanya menunggu.Tabel percepatan: 4 worker memberi 3,8 kali (efisiensi 94%), tetapi 64 worker hanya 28,3 kali (efisiensi 44%): separuh mesin “menganggur” menunggu bagian serial. Pada skenario 100.000 foto, biaya CPU sama (41.667 rupiah) apa pun jumlah mesin, karena jam-mesin totalnya sama; yang berubah adalah lamanya menunggu (13,9 jam lawan 0,9 jam). Catatan: hitungan ini menganggap 100% bisa diparalelkan; pada kenyataannya bagian serial dan beban antrean menambah waktu. Satu mesin GPU dengan tarif tiruan lebih mahal per jam tetapi total lebih murah (25.000 rupiah dalam 2,8 jam) karena foto diproses jauh lebih cepat; asumsi 0,1 detik per foto di GPU adalah angka contoh, bukan hasil ukur.
Sebelum membeli mesin, ukur dulu berapa lama satu tugas dan berapa bagian yang serial. Tugas yang terlalu kecil (beberapa milidetik) justru lebih lambat bila disebar karena ongkos antrean lebih besar daripada kerjanya. Gabungkan tugas kecil menjadi kelompok (batch) sampai sekitar satu detik per tugas.
Proyek Bersambung A5: foto drone dan citra satelit dengan banyak worker
Kita satukan semuanya: bahan 48 berkas, dikerjakan dulu oleh satu proses (acuan), lalu oleh tiga worker Celery, dibandingkan, dan diringkas per petak menjadi CSV.
08_proyek_pipa_citra.py
# Bab 67 - Proyek A5: memproses 48 foto drone dan citra satelit dengan 3 worker, lalu merangkum per petak
import csv
import os
import shutil
import time
from collections import defaultdict
from pathlib import Path
os.environ["ANTREAN"] = "bab67_08"
from citra_tiruan import analisis, buat_citra
from peluncur import matikan, mulai_pekerja, tunggu_siap
from tugas_celery import analisis_terukur, app
kerja = Path(__file__).parent / "_tmp" / "pipa"
shutil.rmtree(kerja, ignore_errors=True)
berkas = buat_citra(kerja / "citra", jumlah_drone=36, jumlah_satelit=12)
print("bahan masukan :", len(berkas), "berkas")
# 1. Acuan: satu proses mengerjakan semuanya berurutan
mulai = time.perf_counter()
hasil_serial = [analisis(b) for b in berkas]
waktu_serial = time.perf_counter() - mulai
# 2. Terdistribusi: tiga worker mengambil tugas dari antrean Redis
proses = mulai_pekerja(jumlah=3, antrean="bab67_08")
nama = []
try:
nama = tunggu_siap(app, 3) # waktu menyalakan worker tidak ikut dihitung
mulai = time.perf_counter()
pesanan = [analisis_terukur.delay(b) for b in berkas]
jawaban = [p.get(timeout=300) for p in pesanan]
waktu_worker = time.perf_counter() - mulai
hasil_worker = [h for h, _ in jawaban]
jumlah_detik_kerja = sum(d for _, d in jawaban) # total waktu kerja semua tugas, diukur di worker
finally:
matikan(proses, app, nama)
print("hasil identik dengan proses tunggal:", hasil_worker == hasil_serial)
# Rata-rata banyaknya tugas yang berjalan bersamaan = total waktu kerja / lama dinding.
# Ukuran ini tahan terhadap komputer yang sedang sibuk; idealnya mendekati 3.
print("rata-rata tugas berjalan serentak lebih dari 1,5:", jumlah_detik_kerja / waktu_worker > 1.5)
# 3. Merangkum per jenis dan petak, lalu menyimpan CSV
kelompok = defaultdict(list)
for h in hasil_worker:
kelompok[(h["jenis"], h["petak"])].append(h)
baris = []
for (jenis, petak), daftar in sorted(kelompok.items()):
baris.append({"jenis": jenis, "petak": petak, "jumlah": len(daftar),
"rata_indeks": round(sum(d["rata"] for d in daftar) / len(daftar), 3),
"persen_hijau": round(sum(d["persen_hijau"] for d in daftar) / len(daftar), 1)})
with open(kerja / "ringkasan.csv", "w", newline="", encoding="utf-8") as f:
penulis = csv.DictWriter(f, fieldnames=list(baris[0]))
penulis.writeheader()
penulis.writerows(baris)
print(f"\n{'jenis':<8} {'petak':<6} {'jumlah':>6} {'rata_indeks':>12} {'persen_hijau':>13}")
for b in baris:
print(f"{b['jenis']:<8} {b['petak']:<6} {b['jumlah']:>6} {b['rata_indeks']:>12.3f} {b['persen_hijau']:>13.1f}")
print("\nurutan petak dari paling rimbun (drone):",
[b["petak"] for b in sorted((x for x in baris if x["jenis"] == "drone"), key=lambda x: -x["rata_indeks"])])bahan masukan : 48 berkas
hasil identik dengan proses tunggal: True
rata-rata tugas berjalan serentak lebih dari 1,5: True
jenis petak jumlah rata_indeks persen_hijau
drone A 12 0.507 51.4
drone B 12 0.378 26.9
drone C 12 0.216 0.4
satelit A 4 0.428 81.2
satelit B 4 0.316 55.9
satelit C 4 0.254 33.4
urutan petak dari paling rimbun (drone): ['A', 'B', 'C']- Hasil dari tiga worker identik dengan hasil satu proses (
True): sistem terdistribusi tidak boleh mengubah jawaban. - Ukuran kecepatan kami pilih agar tahan terhadap komputer yang sedang sibuk: bukan “berapa kali lebih cepat” (yang melompat bila komputer ramai), melainkan rata-rata banyaknya tugas yang berjalan serentak (total waktu kerja diukur di dalam worker dibagi lama dinding). Nilainya di atas 1,5 (idealnya mendekati 3), tanda pekerjaan benar-benar dibagi. Angka kecepatan sebenarnya berbeda di tiap komputer, jadi tidak dicetak.
- Tabel merangkum 48 berkas menjadi enam baris: petak A paling rimbun dan C paling jarang, baik di drone maupun satelit, sesuai cara kita membuat bahan tiruan.
ringkasan.csvsiap dibaca dasbor (Bab 43) atau GIS (Bab 38).
Untuk ribuan citra nyata, perubahannya kecil: ganti buat_citra dengan daftar berkas dari penyimpanan, naikkan jumlah worker (atau pindahkan ke beberapa komputer dengan Redis yang dapat dijangkau semuanya), dan tugas yang berat per citra (misalnya model deteksi pohon dari Bab 56 di GPU) diberi antreannya sendiri.
Latihan
- Pada
03_celery_worker.py, ubah jumlah worker menjadi 1, lalu 5. Baris keluaran mana yang berubah dan mana yang tetap? Mengapa hasil analisis tetap sama persis berapa pun jumlah worker? - Tambahkan tugas Celery baru
ringkas_petak(petak, daftar_hasil)ditugas_celery.pyyang menghitung rata-rata, dan panggil dari skrip. Mengapa tugas itu lebih baik diberi data kecil (daftar hasil), bukan seluruh citra? - Ubah
04_coba_ulang_dan_gagal.pysupaya jumlah ulang maksimum 2 dan tuliskan prediksi jumlah percobaan sebelum menjalankan. - Tantangan: buat antrean kedua bernama
gpu(Bab 66) dan gunakantask_routesCelery agar tugasanalisis_gpuhanya masuk ke antrean itu. Jelaskan di komentar mengapa antrean terpisah penting.
Ringkasan Bab
- Sistem terdistribusi membagi pekerjaan ke banyak worker lewat antrean: produser, broker, worker, backend hasil.
- Antrean sederhana di Redis (
LPUSH/BRPOP) menunjukkan intinya; versi andal memindahkan tugas ke daftar “sedang diproses”. - Celery (banyak fitur) dan RQ (sederhana) sama-sama dijalankan dengan Redis; di Windows kami memakai pool
solo(Celery) danSimpleWorker(RQ), dan Redis 5 memerlukanprotocol=2. - Kegagalan itu biasa: coba-ulang untuk gangguan sesaat, jangan ulang kesalahan data, catat tugas gagal, dan buat tugas idempoten.
- Pantau panjang antrean, status worker, dan tugas gagal.
- Hukum Amdahl membatasi percepatan; biaya CPU relatif tetap sementara waktu tunggu turun; tugas terlalu kecil malah lambat bila disebar.
- Kirim rujukan, bukan data besar, lewat antrean.
Selanjutnya, di Bab 68: penutup perjalanan: menyusun portofolio dari semua proyek Anda, menulis README yang baik, berkontribusi ke proyek terbuka, memilih jalur karier, dan terus belajar.