Belajar Apache Spark - Custom Extensions & UDFs

#Pendahuluan

Bawaan Spark sangat kaya, tapi dunia nyata selalu punya kebutuhan khusus: logika bisnis yang tidak ada di fungsi standar, format data yang tidak dikenal Spark, atau agregasi yang tidak sesuai pola umum. Episode 17 ini membahas bagaimana kalian memperluas Spark — bukan memodifikasinya, melainkan menambahkan kemampuan melalui UDF, UDAF, UDT, dan library kustom.

Menulis extension dengan benar membuat pipeline lebih bersih, karena logika yang sama ditulis sekali dan dipakai di mana-mana. Salah memakainya, sebaliknya, menghasilkan job yang lambat karena melewati optimasi Catalyst.

Episode ini membahas empat area: menulis UDF di Scala, Java, dan Python; UDAF dan UDT; custom serialization dengan Kryo; serta packaging library agar reusable.

#Menulis UDF di Scala, Java, dan Python

#UDF di Python

Python paling mudah untuk memulai. Daftarkan fungsi biasa dengan spark.udf.register atau gunakan decorator:

UDF Python dengan decorator

from pyspark.sql import functions as F
 
@F.udf("string")
def format_gaji(gaji):
    return f"Rp{gaji:,.0f}"
 
df.select(format_gaji(df.gaji).alias("gaji_terformat")).show()

@F.udf("string") mendaftarkan fungsi dengan tipe hasil string. Perlu diingat: UDF Python dijalankan di process Python terpisah dan data dipindah antar JVM dan Python — untuk data besar, biaya ini nyata. Prioritaskan fungsi bawaan atau UDF Scala bila performa kritis.

#UDF di Scala

UDF Scala berjalan di dalam JVM sehingga lebih cepat:

UDF Scala

import org.apache.spark.sql.functions.udf
 
val formatGaji = udf { (gaji: Double) => f"Rp$gaji%,.0f" }
 
df.withColumn("gaji_terformat", formatGaji(df("gaji")))

udf { (gaji: Double) => ... } menciptakan UDF dari lambda. Karena dieksekusi di JVM, tidak ada overhead transfer antar proses — keunggulan utama dibanding UDF Python.

#UDF di Java

Java memakai pendekatan functional interface UDF1<T, R> dan perlu mendaftarkannya secara eksplisit ke SparkSession. Sintaksnya lebih verbose, tapi menghasilkan fungsi yang terintegrasi penuh dengan API Java.

#UDAF dan UDT

#UDAF dengan Aggregator

UDAF menggabungkan banyak baris menjadi satu nilai dengan buffer yang bisa digabung paralel. Implementasi di Scala memakai trait Aggregator:

UDAF dengan Aggregator

import org.apache.spark.sql.Encoder
import org.apache.spark.sql.expressions.Aggregator
 
case class BufferRata2(jumlah: Double, count: Long)
 
object Rata2Tertimbang extends Aggregator[(Double, Double), BufferRata2, Double] {
  def zero: BufferRata2 = BufferRata2(0.0, 0L)
  def reduce(b: BufferRata2, input: (Double, Double)): BufferRata2 =
    BufferRata2(b.jumlah + input._1 * input._2, b.count + input._2.toLong)
  def merge(b1: BufferRata2, b2: BufferRata2): BufferRata2 =
    BufferRata2(b1.jumlah + b2.jumlah, b1.count + b2.count)
  def finish(b: BufferRata2): Double = if (b.count == 0) 0.0 else b.jumlah / b.count
  def bufferEncoder: Encoder[BufferRata2] = Encoders.product
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

reduce menggabungkan baris ke buffer, merge menggabungkan buffer antar partisi, dan finish menghasilkan output akhir. Karena buffer bisa digabung paralel, agregasi tetap berjalan terdistribusi.

#UDT untuk Tipe Kustom

UDT (User-Defined Type) memberitahu Spark cara menyimpan tipe objek khusus dalam kolom. Cocok untuk tipe domain seperti geometri atau vektor khusus. Implementasinya memerlukan UserDefinedType dan pengaturan registrasi — lebih jarang dipakai daripada UDF, tapi penting untuk tipe domain.

#Custom Serialization dan Kryo Registrasi

#Memilih Serializer

Java serialization default aman namun lambat. Kryo jauh lebih cepat dan hemat memori. Aktifkan dan daftarkan class kustom:

spark.serializer   org.apache.spark.serializer.KryoSerializer
spark.kryo.registrationRequired  true
spark.kryo.classesToRegister  com.example.Pegawai,com.example.Pesanan

spark.kryo.classesToRegister mendaftarkan class yang akan diserialisasi. Dengan registrationRequired=true, Spark menolak class yang tidak terdaftar — ini menjaga stabilitas karena format serialisasi tidak berubah diam-diam ketika class baru muncul.

#Mengapa Registrasi Penting

Tanpa registrasi, Kryo memakai pemetaan nama class secara dinamis yang bisa berbeda antar versi library — sumber bug yang sulit dilacak. Registrasi eksplisit membuat format serialisasi stabil dan juga lebih cepat karena ID numerik dipakai menggantikan nama string.

#Packaging dan Reusable Libraries

#Membuat Library dari Kode Kustom

Agar UDF dan UDAF bisa dipakai di banyak job, kumpulkan dalam satu library dan build dengan build tool:

sbt package

Hasilnya berupa jar file. Kirim library bersama job dengan spark-submit:

spark-submit --jars spark-udf-lib.jar job.py

Untuk Scala, gabungkan dependency ke satu jar dengan --packages atau gunakan sbt-assembly untuk fat jar. Pastikan versi library sesuai dengan versi Spark — mismatch versi adalah penyebab klasik error NoSuchMethodError di cluster.

#Mengelola Dependency di Produksi

Praktik yang direkomendasikan:

• Pinned versions: simpan versi dependency di file konfigurasi yang jelas.
• Build reproducibility: gunakan lockfile dan reproducible build (misalnya sbt atau Gradle dengan versi tetap).
• Hindari fat jar untuk library bersama: submit jar sekali lewat --jars lebih baik daripada menggandakan library di setiap job.

#Penutup

Episode 17 memberdayakan kalian untuk memperluas Spark: UDF menambah fungsi per baris di Python, Scala, dan Java; UDAF menciptakan agregasi kustom yang tetap terdistribusi; Kryo mempercepat serialisasi; dan packaging yang rapi membuat library bisa dipakai ulang lintas job.

Inti yang harus dibawa pulang:

• UDF Python sederhana tapi mahal; UDF Scala/JVM jauh lebih cepat.
• UDAF memakai Aggregator dengan reduce, merge, dan finish.
• Kryo lebih cepat dari Java serialization dan wajib didaftarkan.
• Registrasi class Kryo menjaga stabilitas format serialisasi.
• Library kustom di-packaging dan dikirim dengan --jars atau --packages.

Di episode 18 selanjutnya kita akan membahas cross-system integration — menghubungkan Spark dengan Kafka, Cassandra, dan Elasticsearch, peran Spark sebagai ETL engine untuk data lakes dan warehouses, integrasi dengan BI tools, serta pola data ingestion dan CDC.