Distributed Data Processing

Kita sudah tahu dari chapter Spark Partitioning bahwa satu partition diproses oleh satu task. Pertanyaan berikutnya lebih besar: saat pipeline TokoKita berpindah dari laptop ke cluster, siapa yang memutuskan task berjalan di mesin mana, dan bagaimana kita yakin beban terbagi merata? Jawabannya adalah kerja sama tiga pihak — driver, cluster manager, dan executor — plus disiplin memecah data menjadi partition yang seimbang. Pada chapter ini kita menyelami mekanisme penjadwalan task, membuktikan sebaran baris antar task lewat eksekusi nyata, lalu menutup dengan pertanyaan jujur: apakah semua pekerjaan memang perlu cluster.

Driver, Cluster Manager, dan Executor #

Arsitektur cluster Spark terdiri atas tiga peran. Driver adalah proses yang menjalankan kode kita: membangun DataFrame, menyusun execution plan, lalu mengirimkan task. Cluster manager (standalone, YARN, Kubernetes) mengalokasikan sumber daya dan mendaftarkan executor — proses JVM di mesin pekerja yang masing-masing membawa core dan memori sendiri. Setiap executor menyediakan slot; satu slot menjalankan satu task, dan satu task memproses tepat satu partition — itulah aliran kerja lengkap dari konsep partition yang kita kenal.

Penjadwalan task mengikuti prinsip data locality: driver berusaha menempatkan task di executor yang menyimpan partition-nya, karena memindahkan komputasi hampir selalu lebih murah daripada memindahkan data. Ketika satu task ternyata jauh lebih lambat dari saudaranya (misalnya mesinnya sedang sibuk proses lain), Spark dapat menjalankan salinan spekulan task tersebut di executor lain dan mengambil hasil yang selesai lebih dulu — mekanisme speculative execution. Fitur ini nonaktif secara default (spark.speculation) dan hanya aman untuk operasi deterministik: task yang menulis keluaran dengan efek samping tidak boleh dijalankan ganda.

Demo: Sebaran Baris antar Task #

Skrip berikut menulis 100 ribu pesanan TokoKita sebagai Parquet terpartisi per cabang, membacanya kembali sambil mencatat berkas asal tiap baris, lalu memaksa pembagian ulang menjadi 8 partition dan menghitung isi tiap partition:

"""Partisi berkas, partisi input, dan sebaran baris antar task."""
import shutil
from pathlib import Path

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

DATA = Path(__file__).parent / "data_parquet"

spark = (SparkSession.builder
         .master("local[1]")
         .config("spark.ui.enabled", "false")
         .config("spark.sql.shuffle.partitions", "4")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

print("=== Bagian 1: menulis Parquet terpartisi menurut cabang ===")
if DATA.exists():
    shutil.rmtree(DATA)

pesanan = spark.range(100_000).select(
    F.col("id").alias("pesanan_id"),
    (F.col("id") % 8).alias("cabang"),
    F.when(F.col("id") % 3 == 0, F.lit("Budi"))
     .when(F.col("id") % 3 == 1, F.lit("Siti"))
     .otherwise(F.lit("Dewi")).alias("pelanggan"),
    ((F.col("id") % 500) * 1000 + 15000).alias("nilai"),
)
pesanan.write.partitionBy("cabang").parquet(str(DATA))
print("direktori berkas:")
for d in sorted(p.name for p in DATA.iterdir() if not p.name.startswith(".")):
    print(f"  {d}")

print("=== Bagian 2: membaca kembali dan partisi input ===")
df = spark.read.parquet(str(DATA))
print(f"jumlah baris             : {df.count()}")
print(f"jumlah partisi input     : {df.rdd.getNumPartitions()}")
print("berkas asal tiap baris (input_file_name):")
(df.groupBy("cabang", F.regexp_replace(F.input_file_name(), f"^file://{DATA}/", "")
           .alias("berkas"))
   .count().orderBy("cabang")
   .show(truncate=False))

print("=== Bagian 3: sebaran baris setelah repartition ===")
tugas = df.repartition(8)
sebaran = (tugas.withColumn("partisi", F.spark_partition_id())
               .groupBy("partisi").count().orderBy("partisi"))
sebaran.show()

spark.stop()

Output yang Dihasilkan:

=== Bagian 1: menulis Parquet terpartisi menurut cabang ===
direktori berkas:
  _SUCCESS
  cabang=0
  cabang=1
  cabang=2
  cabang=3
  cabang=4
  cabang=5
  cabang=6
  cabang=7
=== Bagian 2: membaca kembali dan partisi input ===
jumlah baris             : 100000
jumlah partisi input     : 1
berkas asal tiap baris (input_file_name):
+------+----------------------------------------------------------------------------+-----+
|cabang|berkas                                                                      |count|
+------+----------------------------------------------------------------------------+-----+
|0     |cabang=0/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|1     |cabang=1/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|2     |cabang=2/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|3     |cabang=3/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|4     |cabang=4/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|5     |cabang=5/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|6     |cabang=6/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
|7     |cabang=7/part-00000-aa44b20d-a642-4196-86eb-ad8b6f35b9b9.c000.snappy.parquet|12500|
+------+----------------------------------------------------------------------------+-----+

=== Bagian 3: sebaran baris setelah repartition ===
+-------+-----+
|partisi|count|
+-------+-----+
|      0|12500|
|      1|12500|
|      2|12500|
|      3|12500|
|      4|12500|
|      5|12500|
|      6|12500|
|      7|12500|
+-------+-----+

Mari kita telaah baris-baris kode di atas:

1. Baris 21-28 (spark.range(100_000).select(...)): membangun pesanan secara round-robin — id % 8 menyebar 100 ribu baris rata ke 8 cabang, 12.500 per cabang.
2. Baris 29 (partitionBy("cabang")): menulis satu direktori per cabang; struktur cabang=0 sampai cabang=7 di output adalah bentuk fisik partisi data di storage.
3. Baris 37 (df.rdd.getNumPartitions()): jumlah partition input yang dibaca Spark dari 8 berkas — hasilnya 1 karena Spark menggabungkan berkas kecil menjadi satu partition yang efisien di mode lokal ini.
4. Baris 39-42 (input_file_name()): melaporkan berkas asal tiap baris, membuktikan cabang 0 memang hanya dibaca dari berkas cabang=0/... — bukti data tiap cabang memang terpisah fisik.
5. Baris 45 (df.repartition(8)): memaksa shuffle menjadi 8 partition; spark_partition_id() di baris 46 memberi nomor partition tiap baris sehingga groupBy("partisi") menghitung isinya.

Membaca Sebaran: Task yang Seimbang #

Bagian 3 output menunjukkan target dari pemrosesan terdistribusi yang sehat: 8 partition berisi 12.500 baris masing-masing. Pada cluster, angka ini berarti 8 task dengan durasi nyaris identik yang bisa berjalan serentak di 8 slot — total durasi menyusut mendekati seperdelapan versi satu task. Sebaliknya, sebaran seperti [88000, 4000, 4000] membuat 2 task selesai cepat lalu menganggur menunggu 1 task yang menanggung 88 ribu baris — fenomena skew yang kita bedah tuntas pada chapter Shuffle dan Skew. Catatan kecil: bagian UUID pada nama berkas parquet (misalnya aa44b20d-...) dibuat acak oleh Spark di setiap penulisan, jadi nilainya berbeda antar run — struktur dan angka barisnya tetap identik.

Cluster atau Satu Mesin #

Kekuatan pemecahan task di atas datang bersama biaya tetap: koordinasi jaringan, serialisasi, dan operasional cluster itu sendiri. Untuk banyak pekerjaan data, satu mesin dengan tool yang tepat masih unggul:

Perbandingan kuantitatifnya — lengkap dengan benchmark jujur yang menunjukkan kapan Spark justru kalah — kita rangkum pada chapter Kapan Menggunakan Spark.

Setiap executor di cluster membawa jatah memori sendiri, dan task yang kehabisan memori tidak gagal begitu saja: Spark menumpahkan data kerjanya ke disk. Mekanisme spill inilah yang kita ukur pada chapter Memory Management.