Scala dengan Apache Spark

Apache Spark adalah mesin analitik data terdistribusi yang API utamanya ditulis dalam Scala, sehingga sebagian besar contoh pekerjaan data di dokumentasi, buku, dan repositori publik berupa kode Scala. Topik ini menjelaskan mengapa Spark 4.x tetap berjalan pada Scala 2.13 meskipun Scala 3 sudah lama dirilis, serta merinci perbedaan sintaks yang terasa ketika membaca kode lintas versi. Pembahasan ditutup dengan proyek scala-cli lengkap yang menghitung frekuensi kata di atas Spark SQL. Setelah membacanya, kode contoh Spark bergaya 2.13 mudah dipahami dan dapat dijalankan dari struktur proyek yang nyata.

Mengapa Spark 4.x berjalan di atas Scala 2.13#

Setiap rilis Spark dikompilasi terhadap satu versi Scala tertentu, dan artefak resminya hanya diterbitkan untuk versi tersebut. Spark 4.x menggunakan Scala 2.13. Catatan rilis Spark 4.0.0 mencatat perubahan yang menghapus Scala 2.12 dan menjadikan Scala 2.13 sebagai basis build, sedangkan build untuk Scala 3 tidak disediakan. Akibatnya, artefak pustaka terbit dengan akhiran _2.13 saja, misalnya org.apache.spark:spark-sql_2.13:4.0.5.

Penyatuan ke satu versi didorong oleh alasan teknis dan organisatoris. Ekosistem Spark bergantung pada kompatibilitas biner (binary compatibility), yaitu jaminan bahwa artefak yang telah dikompilasi tetap dapat digunakan tanpa kompilasi ulang. Jaminan itu harus dijaga terhadap pustaka turunan, seperti konektor sumber data dan pustaka fungsi khusus yang digunakan pada pekerjaan produksi. Membangun Spark untuk dua versi Scala sekaligus berarti menggandakan pekerjaan rilis dan pengujian pada proyek berukuran besar tersebut. Konsolidasi pada Scala 2.13 memangkas beban itu, dan sebagian besar sintaks Scala 2.13 juga sah di Scala 3.

Akibatnya, pekerja data yang membaca kode Spark dalam praktiknya membaca kode Scala 2.13. Contoh program, jawaban forum, dan panduan yang beredar ditulis dalam versi itu. Pemahaman atas sintaks 2.13 karenanya bukan pengetahuan historis, melainkan bekal untuk membaca kode yang masih digunakan secara luas.

Perbedaan sintaks 2.13 yang terasa#

Perbedaan yang terasa ketika berpindah dari Scala 3 ke Scala 2.13 terkonsentrasi pada beberapa titik yang mudah dipetakan. Tabel berikut merangkum titik-titik yang paling sering muncul pada kode Spark.

Dari keempat baris pada tabel, dua yang paling mencolok adalah struktur blok dan wildcard import. Scala 2.13 mewajibkan kurung kurawal untuk badan class, object, metode, dan blok lain, sedangkan Scala 3 menyediakan aturan indentasi sebagai pilihan tambahan. Karena Scala 3 tetap menerima gaya kurawal, kode bergaya 2.13 nyaris selalu terbaca tanpa hambatan oleh pengguna Scala 3, dan itulah titik temu yang menjaga kode lintas versi tetap mudah dibaca.

Sintaks prosedur yang deprecated, yakni bentuk yang masih terkompilasi tetapi diumumkan akan dihapus, ditulis tanpa tipe hasil dan tanpa tanda sama dengan, seperti def main(args: Array[String]) { }. Bentuk itu memunculkan peringatan di 2.13 dan ditolak sebagai error di Scala 3, sehingga kode yang dikelola secara serius selalu menuliskan : Unit =. Wildcard _ pada import, misalnya import spark.implicits._, adalah penanda 2.13 yang paling sering ditemui karena import itu hadir di hampir seluruh contoh DataFrame. Padanannya di Scala 3 adalah import spark.implicits.*, sedangkan argumen berjumlah bebas (varargs, yang banyaknya tidak ditetapkan di awal) ditulis xs: _* di 2.13 dan xs* di Scala 3.

Struktur proyek scala-cli untuk Spark#

Proyek pada topik ini terdiri atas dua berkas yang diletakkan di satu direktori. Berkas pertama, hitung-kata.scala, memuat seluruh program beserta direktif konfigurasi scala-cli di bagian atasnya. Berkas kedua, resources/log4j2.properties, adalah berkas konfigurasi log yang ditempatkan di dalam direktori resources.

Empat direktif di bagian atas berkas Scala menetapkan konfigurasi proyek. Direktif //> using scala 2.13 memilih compiler Scala 2.13 sehingga proyek cocok dengan artefak Spark. Direktif //> using dep org.apache.spark::spark-sql:4.0.5 menambahkan dependensi Spark SQL, dan tanda titik dua ganda memberi tahu scala-cli untuk menambahkan akhiran versi Scala pada nama artefak, sehingga yang diunduh adalah spark-sql_2.13. Direktif //> using javaOpt -Xmx1500m menaikkan batas memori JVM (Java Virtual Machine) karena mode lokal Spark menjalankan mesin eksekusinya di dalam proses JVM yang sama dengan program. Direktif //> using resourceDir ./resources menetapkan direktori resources sebagai sumber resource (berkas pendukung yang disertakan pada classpath, yaitu daftar lokasi tempat JVM mencari kelas dan berkas pendukung), sehingga log4j2.properties ikut dimuat saat aplikasi berjalan.

Spark menuliskan keluaran lognya ke stderr, yaitu kanal keluaran kesalahan standar, melalui kerangka kerja log4j2, sedangkan tabel hasil program dicetak ke stdout. Konfigurasi berikut meredam log sampai level error sehingga layar tidak dibanjiri pesan INFO saat program dijalankan.

# meredam log Spark: hanya level error ke atas yang dicetak
rootLogger.level = error
rootLogger.appenderRef.stdout.ref = console

appender.console.type = Console
appender.console.name = console
appender.console.target = SYSTEM_ERR
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %p %c{1}: %m%n

Program dijalankan dari direktori proyek dengan perintah scala-cli run .. Argumen berupa titik memberi tahu scala-cli untuk memproses seluruh proyek pada direktori tersebut, bukan satu berkas saja. Eksekusi pertama memerlukan waktu lebih lama karena dependensi Spark diunduh terlebih dahulu, sedangkan eksekusi berikutnya cukup cepat karena artefaknya sudah tersimpan di cache.

Membaca program penghitung kata#

Program berikut menghitung frekuensi kata dari tiga kalimat melalui DataFrame API. DataFrame adalah tabel terdistribusi Spark yang tersusun atas kolom bertipe, dan dataset kecilnya dibangun dari Seq berisi tiga kalimat.

//> using scala 2.13
//> using dep org.apache.spark::spark-sql:4.0.5
//> using javaOpt -Xmx1500m
//> using resourceDir ./resources

import org.apache.spark.sql.{SparkSession, functions}

// Menghitung frekuensi kata dari tiga kalimat memakai DataFrame API
object HitungKata {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("HitungKata")
      .master("local[1]")
      .getOrCreate()

    import spark.implicits._

    val kalimat = Seq(
      "pengolah data bekerja pada jadwal piket",
      "pengolah kata menyusun daftar belanja",
      "pengolah data menyimpan catatan sensor suhu"
    ).toDF("kalimat")

    // memecah setiap kalimat menjadi satu baris per kata
    val kata = kalimat.select(
      functions.explode(functions.split($"kalimat", " ")).as("kata")
    )

    // menghitung frekuensi setiap kata, diurutkan dari yang terbanyak
    val jumlah = kata
      .groupBy("kata")
      .count()
      .orderBy($"count".desc, $"kata")

    jumlah.show(false)

    spark.stop()
  }
}

+---------+-----+
|kata     |count|
+---------+-----+
|pengolah |3    |
|data     |2    |
|bekerja  |1    |
|belanja  |1    |
|catatan  |1    |
|daftar   |1    |
|jadwal   |1    |
|kata     |1    |
|menyimpan|1    |
|menyusun |1    |
|pada     |1    |
|piket    |1    |
|sensor   |1    |
|suhu     |1    |
+---------+-----+

Baris SparkSession.builder() adalah pintu masuk ke setiap program Spark. Metode master("local[1]") menetapkan mode lokal dengan satu utas (thread), yaitu bentuk paling sederhana untuk belajar karena seluruh eksekusi berlangsung di dalam satu proses JVM tanpa klaster. Metode getOrCreate() mengembalikan sesi yang sudah ada atau membuat sesi baru apabila belum ada.

Import spark.implicits._ membuka dua kemudahan yang digunakan program, yaitu referensi kolom dengan tanda dolar seperti $"kalimat" dan konversi Seq menjadi DataFrame melalui toDF. Tanpa import tersebut, kedua bentuk tidak terkompilasi. Pola import inilah yang ditemui di hampir semua contoh Spark bergaya 2.13.

Baris-baris transformasi, seperti select, explode, split, groupBy, dan orderBy, tidak langsung menggerakkan data. Spark menyusunnya menjadi rencana eksekusi (execution plan), lalu rencana itu baru dijalankan ketika aksi dipanggil. Metode show(false) adalah aksi pertama pada program, sehingga pemanggilan metode itulah yang memicu penghitungan dan pencetakan tabel. Argumen false menonaktifkan pemotongan kolom sehingga setiap nilai tercetak utuh. Urutan baris pada keluaran terjamin karena orderBy mengurutkan berdasarkan jumlah terbanyak terlebih dahulu, lalu berdasarkan nama kata apabila jumlahnya seri, sehingga keluarannya sama setiap kali program dijalankan. Baris spark.stop() menghentikan sesi dan membuat program berakhir dengan sendirinya.

Mengapa perbedaan versi bukan hambatan besar#

Perbedaan versi paling terasa ketika kode ditulis, bukan ketika kode dibaca. Untuk membaca, pemetaan pada tabel di atas sudah cukup karena selisihnya bersifat mekanis dan setiap bentuk 2.13 memiliki padanan langsung di Scala 3. Untuk menulis, kebanyakan kode bergaya kurawal yang sah di 2.13 lolos compiler Scala 3 tanpa suntingan berarti karena Scala 3 mempertahankan sintaks klasik itu sebagai pilihan resmi. Bagian yang benar-benar perlu disesuaikan hanya sintaks yang sudah dihapus, seperti sintaks prosedur.

Dalam pekerjaan data, bobot yang lebih besar justru berada di API. DataFrame API, fungsi agregasi, dan penanganan tipe data tidak berubah antara kode Scala 2.13 dan Scala 3, sehingga keterampilan yang dibangun pada satu versi terbawa ke versi lainnya. Apabila kelak Spark menerbitkan build untuk Scala 3, penyesuaian sintaks dapat dilakukan dengan bantuan alat migrasi, sedangkan pemahaman konsepnya tidak perlu dibangun ulang. Dengan demikian, membaca kode Spark 2.13 hari ini tetap berguna untuk masa depan ekosistemnya.

Ringkasan#

• Spark 4.x dikompilasi dengan Scala 2.13 dan tidak menerbitkan artefak Scala 3, sehingga kode contoh Spark yang beredar umumnya berupa kode 2.13.
• Perbedaan sintaks yang paling terasa adalah kurung kurawal wajib, sintaks prosedur yang deprecated, serta wildcard _ pada import dan argumen berjumlah bebas.
• Proyek scala-cli untuk Spark menetapkan versi 2.13 melalui direktif, meredam log dengan resources/log4j2.properties, dan dijalankan dengan scala-cli run . dari direktori proyek.

Referensi#

• Spark 4.0.0 Release — Apache Spark — Catatan rilis yang mencatat penghapusan Scala 2.12 dan penetapan Scala 2.13 sebagai basis build.
• Spark SQL Getting Started — Apache Spark — Panduan resmi memulai SparkSession dan DataFrame.
• Spark SQL Programming Guide — Apache Spark — Rujukan resmi transformasi dan aksi pada DataFrame API.