KAL
Menu

SparkSession

Titik masuk, config, dan penamaan aplikasi.

Daftar isi

SparkSession adalah objek yang menjadi titik masuk seluruh pekerjaan PySpark. Topik ini membahas peran SparkSession pada era Spark 2 ke atas, cara membangunnya melalui pola builder, konfigurasi yang paling sering diubah, serta cara menutup sesi dengan benar. Setelah membacanya, sebuah sesi Spark dapat dibuat dengan nama aplikasi yang jelas, dikonfigurasi melalui kunci config yang penting, dan ditutup tanpa meninggalkan proses yang menggantung.

Titik masuk tunggal aplikasi Spark#

Pada era sebelum Spark 2, sebuah program PySpark harus mengelola beberapa objek secara terpisah. Objek SparkContext menangani koneksi ke cluster, SQLContext menangani pemrosesan SQL, dan HiveContext menangani integrasi dengan Hive. Program yang kompleks menjadi penuh objek konteks yang tumpang-tindih, dan tugas-tugas yang mirip bisa dilakukan melalui pintu yang berbeda-beda.

Spark 2 menyatukan ketiganya di bawah SparkSession. Satu objek yang biasanya dinamai spark menjadi pintu masuk untuk semua API: pembuatan DataFrame, eksekusi SQL, akses katalog, dan pembacaan data melalui spark.read. Semua operasi yang diketik pada program driver, yaitu program utama yang berjalan pada mesin tempat aplikasi dimulai, diarahkan melalui objek ini. Di balik layar, SparkSession membungkus satu SparkContext yang aktif, dan sebuah proses JVM hanya mengizinkan satu SparkContext yang aktif pada satu waktu.

Konsistensi satu pintu ini membuat program lebih mudah dibaca dan diuji. Fungsi mana pun dalam program dapat menerima objek spark dan langsung bekerja, tanpa perlu melacak objek konteks mana yang dipakai untuk keperluan tertentu. Konvensi inilah yang dipakai seluruh contoh pada jalur ini.

Membangun sesi dengan pola builder#

SparkSession tidak dibuat dengan constructor langsung, melainkan dirakit melalui pola builder, yaitu rantai pemanggilan metode yang menyusun spesifikasi sesi sebelum sesi benar-benar dibuat. Metode .master("local[2]") menetapkan mode eksekusi lokal dengan dua thread di dalam satu proses JVM. Angka di dalam tanda kurung adalah jumlah thread yang dipakai untuk mengeksekusi task, bukan jumlah proses. Untuk belajar pada satu mesin, mode lokal dengan dua thread sudah mewakili perilaku dasar paralelisme Spark.

Metode .appName("nama-aplikasi") memberi nama pada aplikasi. Nama ini muncul pada antarmuka pemantauan Spark UI dan pada log cluster, sehingga aplikasi yang berjalan bersama aplikasi lain tetap mudah dibedakan. Penamaan yang deskriptif membantu saat mencari jejak aplikasi pada log produksi.

Metode .getOrCreate() mengakhiri rantai builder. Metode ini mengembalikan sesi yang sudah ada apabila tersedia, dan membuat sesi baru apabila belum ada. Perilaku ini penting karena satu proses JVM hanya boleh memiliki satu SparkContext aktif, sehingga pemanggilan builder berulang kali pada proses yang sama tetap mengembalikan sesi yang sama, bukan sesi tambahan.

Config yang sering diubah#

Konfigurasi diset melalui .config("kunci", "nilai") sebelum .getOrCreate() dipanggil. Dua kunci yang paling berguna untuk dikenal sejak awal adalah spark.sql.shuffle.partitions dan spark.sql.adaptive.enabled. Kunci pertama menetapkan jumlah partisi hasil shuffle, yaitu proses pengumpulan ulang data lintas partisi saat operasi seperti orderBy dijalankan. Nilai default-nya adalah 200, angka yang berlebihan untuk data berukuran kecil, sehingga contoh di jalur ini sering menurunkannya.

Kunci kedua mengaktifkan Adaptive Query Execution (AQE), mekanisme Spark yang menyetel ulang rencana eksekusi secara otomatis ketika job sedang berjalan, misalnya dengan menggabungkan partisi kecil. Pada Spark 4 kunci ini bernilai aktif secara default, sehingga partisi hasil shuffle yang berukuran kecil digabungkan secara otomatis ketika job berjalan. Untuk mengubah config saat sesi sudah berjalan, gunakan spark.conf.set seperti pada fragmen berikut.

python 
# tidak-dijalankan
# mengubah config saat sesi sudah berjalan
spark.conf.set("spark.sql.shuffle.partitions", "16")

# membaca nilai config
print(spark.conf.get("spark.sql.shuffle.partitions"))

Skrip pertama: membuat, memeriksa, menutup#

Skrip berikut merakit sesi lengkap, memeriksa versi Spark dan beberapa nilai config, lalu menutup sesi secara eksplisit. Skrip disimpan sebagai pengenal_sparksession.py dan dijalankan dari direktori topik.

python pengenal_sparksession.py 
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("pengenal-sparksession")
    .config("spark.sql.shuffle.partitions", "8")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

print("versi:", spark.version)
print("master:", spark.conf.get("spark.master"))
print("nama aplikasi:", spark.conf.get("spark.app.name"))
print("shuffle partitions:", spark.conf.get("spark.sql.shuffle.partitions"))
print("adaptive:", spark.conf.get("spark.sql.adaptive.enabled"))

spark.stop()
print("sesi ditutup dengan spark.stop()")
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
versi: 4.2.1
master: local[2]
nama aplikasi: pengenal-sparksession
shuffle partitions: 8
adaptive: true
sesi ditutup dengan spark.stop()

Keluaran menunjukkan bahwa setiap nilai yang diset melalui builder benar-benar berlaku. Properti spark.master berisi local[2] sesuai argumen .master(), dan spark.app.name berisi nama yang diberikan kepada .appName(). Nilai spark.sql.shuffle.partitions mencetak 8, bukan default 200, karena skrip menetapkannya melalui .config(). Baris pertama, yang berupa peringatan JVM tentang incubator modules, berasal dari lingkungan Java dan tidak berhubungan dengan kode skrip.

Metode spark.stop() menghentikan sesi, melepaskan resource yang dipakai, dan menutup koneksi ke cluster. Setelah dipanggil, objek spark tidak dapat digunakan lagi pada proses tersebut. Program yang selesai bekerja sebaiknya selalu menutup sesinya secara eksplisit agar perilaku berakhirnya program tidak bergantung pada mekanisme pembersihan otomatis.

Kesalahan umum#

Kesalahan pertama adalah memanggil rantai builder tanpa .getOrCreate(). Rantai .master().appName().config() hanya merakit spesifikasi, dan tidak ada sesi yang dibuat sampai .getOrCreate() dipanggil. Hasil rantai tersebut berupa objek builder, bukan sesi, sehingga pemanggilan metode milik sesi seperti spark.sql atau spark.read padanya berhenti dengan error AttributeError.

Kesalahan kedua adalah menafsirkan local[2] sebagai dua proses atau dua mesin. Mode lokal berjalan sepenuhnya di satu proses JVM, dan angka 2 hanya menyatakan dua thread eksekusi. Perkiraan yang salah membuat pengamatan tentang pemakaian CPU menjadi keliru, misalnya mengira dua proses Java seharusnya muncul padahal yang bekerja hanyalah satu proses dengan dua thread.

Kesalahan ketiga adalah meninggalkan sesi tanpa spark.stop(). Pada program singkat akibatnya nyaris tidak terasa, tetapi pada aplikasi yang berjalan lama sesi yang tidak ditutup membiarkan pekerjaan latar dan pemantauan tetap hidup sehingga memori driver tidak pernah dilepas. Biasakan menutup sesi pada akhir skrip, termasuk pada jalur penanganan error.

Ringkasan#

  • SparkSession menyatukan peran SparkContext, SQLContext, dan HiveContext era sebelum Spark 2 menjadi satu titik masuk bernama spark.
  • Sesi dibangun melalui pola builder: .master("local[2]") berarti dua thread pada satu proses JVM, .appName() menamai aplikasi, dan .getOrCreate() membuat atau mengembalikan sesi.
  • Config penting diset melalui .config() sebelum sesi dibuat, misalnya spark.sql.shuffle.partitions dan spark.sql.adaptive.enabled, dan dapat dibaca dengan spark.conf.get.
  • spark.stop() menutup sesi dan melepaskan resource secara eksplisit.

Referensi#