Belajar Apache Spark - Performance Tuning & Optimization

#Pendahuluan

Kode Spark yang benar belum tentu cepat. Di episode 9 ini kalian akan belajar bahwa performa Spark sangat ditentukan oleh konfigurasi dan pemahaman terhadap cara engine bekerja. Dua query yang menghasilkan output sama bisa berbeda puluhan kali lipat durasinya tergantung bagaimana Spark diminta mengeksekusinya.

Mengapa ini penting? Karena di produksi, resource cluster tidak gratis dan waktu adalah biaya. Sebuah job ETL yang berjalan dua jam alih-alih dua puluh menit berarti pemborosan besar. Engineer yang mampu mengoptimasi Spark bukan hanya menghemat uang, tapi juga memperlambat waktu tunggu seluruh tim data.

Episode ini membahas empat pilar optimasi: tuning konfigurasi dasar, broadcast join dan caching, membaca execution plan Catalyst, serta strategi menghadapi data skew dan shuffle overload.

#Tuning Konfigurasi Dasar

#Konfigurasi Memory

Memori executor terbagi menjadi beberapa wilayah. Dua yang paling penting: execution memory untuk operasi seperti join dan shuffle, dan storage memory untuk cache. Keduanya berbagi satu pool yang bisa saling meminjam, dikontrol oleh spark.memory.fraction:

spark-submit --executor-memory 8g --conf spark.memory.fraction=0.6 job.py

Aturan praktis: sisakan memori untuk JVM overhead dan sistem. Jika executor kehabisan memori, Spark akan memindahkan (spill) data ke disk dan job melambat drastis. Pastikan --executor-memory seimbang dengan jumlah core agar tidak membayar resource yang menganggur.

#Konfigurasi Shuffle

Shuffle adalah biaya terbesar. Dua parameter yang sering diatur:

• spark.sql.shuffle.partitions: jumlah partisi output setelah shuffle — default 200.
• spark.shuffle.compress: kompresi data shuffle — biarkan true.

spark-submit --conf spark.sql.shuffle.partitions=100 job.py

Nilai ideal bergantung pada jumlah core dan ukuran data. Terlalu sedikit membuat tiap task berat, terlalu banyak membuat banyak task kecil yang overhead-nya lebih besar dari manfaatnya.

#Paralelisme

Paralelisme ditentukan oleh jumlah partisi, bukan jumlah core. Aturan umum: targetkan 2-3 partisi per core untuk workload CPU-bound, dan lebih banyak lagi untuk workload I/O karena task akan menunggu jaringan atau disk. Inspeksi paralelisme berjalan dengan:

Melihat jumlah partisi

df.rdd.getNumPartitions()

#Broadcast Join dan Caching

#Broadcast Join

Jika satu sisi join berukuran kecil (biasanya di bawah 10MB default, bisa diset sampai 200MB dengan autoBroadcastJoinThreshold), Spark bisa mengirim salinan ke seluruh executor. Dengan begitu, join dilakukan secara lokal tanpa shuffle sama sekali:

Broadcast join eksplisit

from pyspark.sql import functions as F
 
dimensi = spark.read.parquet("data/dim_kota.parquet")
hasil = transaksi.join(F.broadcast(dimensi), "kota_id")

F.broadcast(dimensi) memberi tahu Catalyst untuk memperlakukan dimensi sebagai broadcast relation. Pada join yang membandingkan tabel transaksi raksasa dengan tabel dimensi kecil, teknik ini bisa memangkas waktu hingga puluhan kali lipat.

#Caching dan Persistence

Data yang dipakai berulang kali dalam satu aplikasi sebaiknya di-cache agar tidak dihitung ulang dari sumber:

Cache dan persist

df.persist()
print(df.count())
print(df.filter("stok > 0").count())
df.unpersist()

Tanpa persist(), setiap action memicu rekalkulasi penuh dari sumber. Namun ingat: caching memakan storage memory. Cache hanya data yang benar-benar dipakai ulang, dan panggil df.unpersist() saat selesai agar memori dilepas.

#Partition Tuning

Saat partisi tidak seimbang atau terlalu banyak, sesuaikan dengan repartition() atau coalesce():

Mengubah jumlah partisi

df = df.repartition(200, "kota_id")
df = df.coalesce(8).write.parquet("data/hasil.parquet")

repartition memicu shuffle penuh dan berguna saat menambah paralelisme; coalesce menggabungkan partisi tanpa shuffle penuh dan berguna saat mengecilkan jumlah file sebelum menulis.

#Catalyst dan Execution Plan

#Membaca Physical Plan

Optimasi terbaik dimulai dari diagnosis. df.explain() menampilkan bagaimana Spark akan mengeksekusi query:

Melihat rencana eksekusi

df.groupBy("kota").agg(F.sum("jumlah")).explain("extended")

Output menampilkan parsed logical plan, analyzed plan, optimized plan, hingga physical plan. Perhatikan tiga hal saat membacanya:

• ScanFilter: apakah filter diterapkan sedini mungkin di sumber data.
• Exchange: node yang menandakan shuffle — semakin sedikit, semakin baik.
• BroadcastHashJoin vs SortMergeJoin: strategi join yang dipilih Catalyst.

#Optimasi yang Sudah Otomatis

Catalyst melakukan banyak hal tanpa campur tangan kalian:

• Predicate pushdown: filter dibawa ke sumber data sehingga file yang dibaca lebih sedikit.
• Projection pruning: kolom yang tidak dipakai tidak pernah dibaca.
• Join reordering: urutan join dipilih berdasarkan statistik.
• Codegen: menghasilkan bytecode JVM khusus untuk query tertentu.

Tugas kalian bukan menggantikan Catalyst, tapi memastikan data memberi sinyal yang benar — misalnya menyediakan statistik tabel yang akurat agar join reordering bekerja baik.

#Menghindari Skew, Shuffle Overload, dan Operasi Mahal

#Data Skew

Data skew terjadi ketika satu partisi menangani jauh lebih banyak data daripada yang lain — misalnya satu kota dengan 90 persen transaksi. Gejalanya: satu task memakan waktu lama sementara task lain selesai cepat. Solusi umum:

• Salting: tambahkan prefix acak pada kunci panas untuk menyebar beban.
• Broadcast join: untuk kunci panas dengan sisi kecil.
• Adaptive Query Execution (AQE): aktifkan spark.sql.adaptive.enabled=true yang secara otomatis menangani skew join di Spark 3+.

spark-submit --conf spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.skewJoin.enabled=true job.py

#Shuffle Overload

Setiap join, groupBy, dan reduceByKey memicu shuffle. Untuk mengurangi beban:

• Ganti groupByKey dengan reduceByKey atau agregasi partial.
• Gunakan broadcast join untuk sisi kecil.
• Filter dan pangkas kolom sebelum join, bukan sesudahnya.
• Hindari collect() pada data besar — ini mengirim semua data ke driver dan mematikan memori driver.

#Operasi yang Mahal

Beberapa operasi perlu dihindari atau didesain ulang:

collect()     → kirim semua data ke driver
cartesian()   → meledak secara eksponensial
groupByKey()  → shuffle semua nilai mentah
udf pada baris → tanpa pushdown, jauh lebih lambat dari ekspresi Spark

Khusus untuk UDF: jika bisa diekspresikan dengan fungsi bawaan pyspark.sql.functions, pakai itu. UDF Python memaksa data melintasi batas JVM-Python dan menghilangkan optimasi Catalyst.

#Penutup

Episode 9 menuntaskan keterampilan optimasi: tuning memori dan shuffle menentukan batas resource, broadcast join dan caching mengurangi pekerjaan berulang, membaca physical plan adalah keterampilan diagnosis, dan menghindari skew serta shuffle overload adalah seni menjaga pipeline tetap seimbang.

Inti yang harus dibawa pulang:

• Mulai tuning dari pengamatan, bukan tebakan: gunakan explain() dan Spark UI.
• Broadcast join adalah senjata pertama untuk join dengan sisi kecil.
• Cache hanya data yang dipakai ulang, dan lepaskan dengan unpersist().
• Aktifkan Adaptive Query Execution untuk menangani skew otomatis.
• Hindari collect(), cartesian(), dan groupByKey pada data besar.

Di episode 10 selanjutnya kita akan membahas Spark Streaming dan Structured Streaming — perbedaan DStream versus Structured Streaming, membangun pipeline streaming dengan readStream dan writeStream, trigger modes, watermarking, dan output modes, serta sink ke Kafka, file, console, dan storage.