Belajar Data Engineer - Spark Distributed Processing

#Pendahuluan

Di episode 10 kalian menjalankan PySpark untuk batch processing — dan mungkin bertanya-tanya: kenapa job saya lambat padahal datanya "cuma" 10 GB? Jawabannya hampir selalu bukan kapasitas, melainkan cara Spark memecah dan memindahkan data. Episode 21 membawa kalian ke dalam cara kerja mesin: partitioning, shuffle, caching, dan join strategy.

Mengapa ini penting? Karena di production, performa Spark menentukan biaya cloud. Job yang lambat berarti cluster lebih besar, lebih lama, lebih mahal — dan data engineer yang mengerti optimasi adalah orang yang menghemat ratusan juta rupiah per bulan.

#RDD vs DataFrame

Spark punya dua API utama. Memahami perbedaannya menjelaskan banyak keputusan optimasi:

Aturan utama: selalu pakai DataFrame. RDD hanya untuk kasus khusus yang benar-benar butuh kontrol rendah. DataFrame memanfaatkan Catalyst optimizer — Spark menyusun rencana eksekusi terbaik untuk query kalian secara otomatis.

#Partitioning: Bagaimana Data Terbagi

Spark memproses data dalam partisi — potongan data yang diproses paralel oleh executor. Jumlah partisi menentukan paralelisme:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
 
spark = SparkSession.builder \
    .appName("partition_opt") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()
 
df = spark.read.parquet("s3://data-lake/bronze/orders/")
print(f"Partisi saat dibaca: {df.rdd.getNumPartitions()}")
 
# Repartisi untuk paralelisme lebih tinggi
df2 = df.repartition(400)
 
# Coalesce: mengurangi partisi tanpa shuffle (hanya untuk menurunkan)
df3 = df.coalesce(8)

Prinsipnya:

• Terlalu sedikit partisi (kurang dari executor × cores) → resource menganggur.
• Terlalu banyak partisi kecil → overhead scheduling mengalahkan manfaat paralelisme.
• Aturan praktis: sekitar 2-4x jumlah cores, atau target 100-200 MB per partisi.

#Shuffle: Operasi Paling Mahal di Spark

Shuffle terjadi ketika Spark harus mengelompokkan data dari semua partisi — misalnya pada groupBy, join, atau distinct. Data dipindahkan antar executor melalui jaringan, dan ini operasi paling mahal di Spark.

Cara mengurangi shuffle:

1. Hindari shuffle yang tidak perlu — filter sebelum join/groupBy, bukan sesudahnya.
2. Tambah partisi hanya bila perlu — repartition adalah shuffle itu sendiri; coalesce (tanpa shuffle) untuk menurunkan.
3. Pakai bucketing untuk tabel yang sering di-join dengan key yang sama — data yang akan di-join sudah co-located.

# Lambat: kelompokkan semua dulu, baru filter
slow = df.groupBy("customer_id").count().filter(col("count") > 100)
 
# Cepat: filter dulu, barulah groupBy (lebih sedikit data di-shuffle)
fast = df.filter(col("status") == "completed") \
         .groupBy("customer_id").count() \
         .filter(col("count") > 100)

Perhatikan prinsip universalnya: perkecil data sebelum operasi mahal.

#Caching: Hemat Ulang Tanpa Baca Ulang

Jika sebuah DataFrame dipakai berulang kali dalam job yang sama (misalnya dimension table yang di-join ke banyak fact), baca ulang dari disk adalah pemborosan. Cache menyimpan hasil di memori:

dim = spark.read.parquet("s3://data-lake/gold/dim_customer/")
dim.cache()          # atau persist(MEMORY_AND_DISK)
dim.count()          # force action agar benar-benar di-cache
 
result1 = fact.join(dim, "customer_id").groupBy("segment").sum()
result2 = fact2.join(dim, "customer_id").groupBy("city").sum()

Perbedaan cache() vs persist():

• cache() = persist(MEMORY_AND_DISK) — memori dulu, spill ke disk jika penuh.
• persist(level) — kontrol level: MEMORY_ONLY (cepat, boros memori), DISK_ONLY (hemat memori, lambat).

Aturan praktis: cache hanya jika dipakai lebih dari satu kali dan hasilnya kecil-menengah. Meng-cache DataFrame raksasa malah memicu eviction dan memperlambat.

#Broadcast Join: Join dengan Tabel Kecil

Join biasa (shuffle join) memindahkan kedua sisi berdasarkan key. Jika satu sisi kecil (dimension table, misal di bawah beberapa ratus MB), broadcast join menyalin sisi kecil ke setiap executor — menghilangkan shuffle total:

from pyspark.sql.functions import broadcast
 
# Sisi kecil (< 10MB default threshold) → broadcast otomatis jika memungkinkan
small = spark.read.parquet("s3://data-lake/gold/dim_customer/")
 
# Eksplisit memaksa broadcast
result = fact.join(
    broadcast(small), "customer_id", "inner"
)

Bandingkan cost: shuffle join memindahkan kedua dataset lewat jaringan; broadcast join mengirim satu kali salinan tabel kecil ke semua executor. Untuk join fact (milliaran baris) × dimension (jutaan baris), perbedaan ini bisa mengubah 40 menit menjadi 4 menit.

Tingkatkan threshold hanya jika yakin tabel kecil cukup muat di memori executor:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100 MB

#Praktik: Mendiagnosis Job yang Lambat

Workflow debugging job Spark yang lambat:

1. Buka Spark UI (port 4040) → cek stage yang paling lama.
2. Lihat apakah ada shuffle (Stage description mengandung "shuffle read/write").
3. Cek skew: beberapa task jauh lebih lama dari task lain dalam stage yang sama.
4. Terapkan perbaikan berdasarkan diagnosis:

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, col
 
spark = SparkSession.builder \
    .config("spark.sql.shuffle.partitions", "400") \
    .getOrCreate()
 
fact = spark.read.parquet("s3://data-lake/bronze/orders/")
dim = spark.read.parquet("s3://data-lake/gold/dim_customer/").cache()
dim.count()
 
# Filter dulu sebelum join & agregasi
fact_filtered = fact.filter(col("status") == "completed")
 
# Broadcast join dengan dimensi kecil
joined = fact_filtered.join(broadcast(dim), "customer_id", "inner")
 
# Agregasi dengan partisi cukup
result = joined.groupBy("segment").count()
result.write.mode("overwrite").parquet("s3://data-lake/gold/segment_counts/")

#Kesalahan Umum (Common Pitfalls)

1. .collect() di job produksi. Memindahkan seluruh data ke driver = OOM. collect hanya untuk hasil kecil & debugging.

2. Meng-cache tanpa count(). Cache bersifat lazy — tanpa action, tidak ada yang tersimpan.

3. Broadcast join untuk tabel besar. Jika "tabel kecil" tak muat di memori executor, broadcast malah OOM. Ukur dulu.

4. Menggunakan RDD untuk segala. Python RDD jauh lebih lambat dari DataFrame. Gunakan DataFrame kecuali ada alasan kuat.

#Penutup

Di episode 21 ini kalian telah menguasai optimasi Spark:

• RDD vs DataFrame: selalu DataFrame — Catalyst optimizer mengoptimasi otomatis.
• Partitioning menentukan paralelisme; target ~100-200 MB per partisi.
• Shuffle adalah operasi termahal — perkecil data sebelum shuffle, hindari shuffle tak perlu.
• Caching untuk data yang dipakai berulang; broadcast join menghilangkan shuffle untuk sisi kecil.
• Praktik: diagnosis via Spark UI → temukan stage lambat → terapkan perbaikan tepat.

Di episode 22 selanjutnya kita menyatukan performa dan biaya: performance & cost optimization — query tuning, clustering & partitioning, serverless vs provisioned, dan FinOps data, plus praktik mengurangi cost pipeline. Sampai jumpa di episode 22!