Belajar Apache Spark - Dataset API & Strong Typing

#Pendahuluan

DataFrame memberi kemudahan, tapi kadang kalian butuh jaminan tipe data pada saat kompilasi — bukan saat runtime. Di sinilah Dataset berperan: abstraksi typed yang hanya tersedia penuh di Scala dan Java.

Episode 6 ini membahas Dataset secara mendalam: apa itu Dataset, bagaimana encoders memberikan type safety, perbedaan struktural dengan DataFrame, dan bagaimana berpindah antar keduanya. Materi ini paling relevan bagi kalian yang menulis Spark di Scala atau Java, dan tetap berguna sebagai pengetahuan arsitektural bagi pemakai PySpark.

#Penggunaan Dataset di Scala dan Java

#Konsep Dasar Dataset

Dataset adalah kumpulan terdistribusi dari objek bertipe. Di Scala, sebuah Dataset[T] menyimpan baris sebagai objek dari tipe T — bisa berupa case class, tuple, atau tipe primitif. Di Java, tipe-nya adalah Dataset<T> dengan Java Bean atau Encoders untuk tipe tertentu.

Dataset dari case class di Scala

import org.apache.spark.sql.SparkSession
 
case class Pegawai(nama: String, gaji: Double)
 
val spark = SparkSession.builder.master("local[*]").appName("dataset").getOrCreate()
import spark.implicits._
 
val ds: Dataset[Pegawai] = Seq(
  Pegawai("budi", 8000000),
  Pegawai("sari", 9500000)
).toDS()
 
ds.filter(_.gaji > 8500000).show()

Perhatikan ds.filter(_.gaji > 8500000) — di sini _ merujuk pada objek Pegawai utuh, bukan sebuah Row. Kompiler memeriksa bahwa gaji adalah field yang valid dari case class, sehingga kesalahan nama field tertangkap saat kompilasi, bukan di tengah cluster.

#Dataset di Java

Di Java, Dataset bekerja dengan class yang memenuhi JavaBean convention:

Dataset<Pegawai> ds = spark.read().csv("data/pegawai.csv")
    .as(Encoders.bean(Pegawai.class));

Encoders.bean(Pegawai.class) memberi tahu Spark cara mengonversi data ke objek Pegawai. Sintaksnya lebih panjang daripada Scala, tapi jaminan type safety yang sama tetap berlaku.

#Encoders dan Compile-Time Type Safety

#Apa Itu Encoder

Encoder adalah mekanisme yang mengubah objek JVM menjadi representasi internal Spark dan sebaliknya. Setiap tipe data punya encoder: case class diberi encoder otomatis melalui spark.implicits, sementara tipe khusus perlu Encoders.* eksplisit.

Encoder eksplisit untuk tipe khusus

import org.apache.spark.sql.Encoders
 
val ds: Dataset[Pegawai] = spark
  .read
  .option("header", "true")
  .csv("data/pegawai.csv")
  .as(Encoders.product[Pegawai])

Dengan Encoders.product[Pegawai], Spark memvalidasi bahwa kolom di file cocok dengan field case class. Perbedaan tipe seperti kolom string versus field Double akan melaporkan error dengan jelas.

#Keuntungan Type Safety

Manfaat nyata dari type safety:

• Error lebih awal: kesalahan field atau tipe tertangkap saat kompilasi.
• Refactoring aman: mengganti nama field case class langsung menandai semua pemakaian yang salah.
• Kode yang mudah dibaca: pipeline terlihat seperti manipulasi koleksi Scala biasa.

#Struktur Dataset vs DataFrame

#Hubungan Keduanya

Secara arsitektur, DataFrame adalah alias dari Dataset[Row]. Di Spark Scala API, DataFrame sebenarnya adalah tipe alias — tidak ada kelas terpisah. Ini menjelaskan mengapa semua operasi DataFrame tersedia untuk Dataset dan sebaliknya.

DataFrame = Dataset[Row]   (baris bertipe Row, tanpa tipe khusus)
Dataset[T] = baris bertipe objek T (type-safe)

#Kapan Memilih Mana

• DataFrame: untyped, cocok untuk data tanpa model objek, integrasi SQL, dan eksplorasi cepat. API-nya memakai string nama kolom.
• Dataset: typed, cocok untuk pipeline yang butuh maintainability jangka panjang, domain object, dan pemeriksaan kompilasi.

Tidak ada jawaban tunggal yang benar. Banyak tim memakai DataFrame di lapisan ingestion dan eksplorasi, lalu beralih ke Dataset untuk lapisan inti transformasi bisnis.

#Trade-off Performa

Kedua API dikompilasi ke physical plan yang sama setelah melewati Catalyst, sehingga perbedaan performa umumnya kecil. Namun operasi berorientasi Row (seperti map dan flatMap pada Dataset) bisa lebih lambat daripada operasi deklaratif karena melibatkan serialisasi objek. Untuk transformasi sederhana, pakai ekspresi Spark daripada map Scala bila memungkinkan.

#Konversi Antara DataFrame dan Dataset

#Dari DataFrame ke Dataset

Konversi dilakukan dengan .as[T]:

Konversi DataFrame ke Dataset

val df = spark.read.option("header", "true").csv("data/pegawai.csv")
val ds: Dataset[Pegawai] = df.as[Pegawai]

df.as[Pegawai] memvalidasi kesesuaian schema kolom dengan field case class. Jika ada kolom yang tidak cocok, error muncul segera.

#Dari Dataset ke DataFrame

Sebaliknya, cukup panggil toDF():

Konversi Dataset ke DataFrame

val dfKembali = ds.toDF()

ds.toDF() mengubah objek typed kembali menjadi Dataset[Row]. Ini berguna saat kalian ingin menulis hasil dengan SQL atau menyimpannya ke tabel yang membutuhkan representasi Row.

Dataset[T] .as[T] dari DataFrame   ← df.as[Pegawai]
DataFrame  toDF dari Dataset       ← ds.toDF()

#Penutup

Episode 6 membuka lapisan typed Spark: Dataset memberi type safety saat kompilasi melalui encoders, tersedia penuh di Scala dan Java, dan secara arsitektur DataFrame hanyalah Dataset[Row]. Konversi dua arah membuat kedua API bisa dipakai berdampingan sesuai kebutuhan.

Inti yang harus dibawa pulang:

• Dataset adalah koleksi terdistribusi dari objek bertipe T.
• DataFrame adalah alias dari Dataset[Row] — bukan kelas terpisah.
• Encoders menangani konversi objek JVM dan memberikan type safety.
• Kesalahan tipe tertangkap saat kompilasi, bukan di tengah cluster.
• Konversi lewat .as[T] dan .toDF() bersifat dua arah dan murah.

Di episode 7 selanjutnya kita akan membahas joins, aggregations, dan window functions — semua tipe join dari inner sampai anti, agregasi dengan groupBy, fungsi window seperti row_number dan lead, serta implikasi performa dari join dan shuffle yang menjadi penyebab utama query lambat.