KAL
Menu

Spark SQL

Temp view dan menjalankan SQL pada DataFrame.

Daftar isi

Spark SQL adalah mesin SQL yang tertanam di dalam Spark, dan spark.sql() adalah pintu masuknya. Dengan mendaftarkan sebuah DataFrame sebagai view sementara, seluruh perintah SQL yang sudah umum dikenal, mulai dari SELECT, WHERE, sampai GROUP BY, dapat dijalankan langsung terhadap data yang sedang diproses. Topik ini menunjukkan pola dasar tersebut pada satu berkas CSV kecil, mulai dari mendaftarkan view, menjalankan kueri, sampai menggabungkan hasil SQL kembali ke alur DataFrame.

Mendaftarkan DataFrame sebagai temp view#

View sementara atau temp view adalah nama yang menunjuk ke sebuah DataFrame dalam sesi yang sedang berjalan, bukan tabel fisik yang ditulis ke penyimpanan. Tidak ada data yang disalin ketika view dibuat, karena view hanyalah pemetaan nama ke rencana logis DataFrame. Setiap kali SQL merujuk nama tersebut, Spark menyusun ulang rencana yang sama seolah kueri ditulis langsung dengan API DataFrame.

Fungsi yang dipakai adalah createOrReplaceTempView. Nama fungsi itu sendiri menggambarkan perilakunya, yaitu membuat view baru atau mengganti yang sudah ada tanpa error. Apabila view lama tidak boleh tertimpa secara diam-diam, fungsi createTempView yang lebih ketat akan memunculkan AnalysisException ketika nama sudah terpakai. Kedua view ini hidup selama sesi hidup, dan hilang begitu sesi berakhir, sehingga cocok untuk pekerjaan analisis dalam satu skrip.

Menjalankan SQL pada data siswa#

Skrip berikut membaca berkas nilai_ujian.csv yang berisi enam siswa dari dua kelas, mendaftarkannya sebagai view bernama nilai_ujian, lalu menjalankan dua kueri. Kueri pertama menyaring siswa dengan nilai minimal 88, dan kueri kedua merangkum jumlah serta rata-rata nilai per kelas.

python uji_sql.py 
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("uji-sql")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

df = spark.read.option("header", "true").csv("nilai_ujian.csv")
df.createOrReplaceTempView("nilai_ujian")

print("=== SELECT dasar ===")
spark.sql("SELECT nama, kelas, nilai FROM nilai_ujian WHERE nilai >= 88 ORDER BY nilai").show()

print("=== GROUP BY ===")
spark.sql("""
    SELECT kelas, count(*) AS jumlah, avg(nilai) AS rata_rata
    FROM nilai_ujian
    GROUP BY kelas
    ORDER BY kelas
""").show()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== SELECT dasar ===
+-----+-----+-----+
| nama|kelas|nilai|
+-----+-----+-----+
|  Eko|  10A|   88|
| Budi|  10A|   90|
| Dewi|  10B|   92|
|Farah|  10B|   95|
+-----+-----+-----+

=== GROUP BY ===
+-----+------+-----------------+
|kelas|jumlah|        rata_rata|
+-----+------+-----------------+
|  10A|     3|87.66666666666667|
|  10B|     3|88.33333333333333|
+-----+------+-----------------+

Kedua kueri berjalan persis seperti SQL pada basis data relasional biasa. Perhatikan satu perilaku yang mudah terlewat, yaitu kolom nilai dibaca dari CSV sebagai string, tetapi perbandingan nilai >= 88 tetap berhasil karena Spark mengonversi string ke angka secara implisit pada ekspresi perbandingan. Kelaziman seperti ini memudahkan eksplorasi cepat, tetapi pada data sungguhan schema yang jelas dianjurkan, sebagaimana dibahas pada topik Schema di modul SparkSession.

Hasil GROUP BY menunjukkan dua kelas dengan tiga siswa masing-masing dan rata-rata yang berbeda tipis. Angka 87.66666666666667 pada kolom rata_rata adalah bilangan pecahan presisi ganda yang dicetak apa adanya, dan pembulatan tampilan bisa dilakukan di SQL dengan fungsi round() apabila laporan memang memerlukannya.

Kapan SQL dipilih di dalam skrip Python#

Meskipun penulisan jalur ini berbasis Python, spark.sql() layak dipakai ketika logika yang ditulis lebih alami dinyatakan dalam bentuk tabel. Transformasi berlapis dengan JOIN dan jendela analitik sering kali lebih mudah dibaca sebagai satu pernyataan SQL daripada rantai fungsi yang panjang. Tim yang sudah menyimpan banyak kueri SQL lama juga dapat menggunakannya kembali tanpa menerjemahkan setiap baris ke API DataFrame.

Dua pola pencampuran sering muncul dalam pekerjaan nyata. Pola pertama adalah menghitung bagian berat dengan SQL lalu melanjutkan hasilnya sebagai DataFrame, misalnya hasil = spark.sql("SELECT ... FROM nilai_ujian") diikuti hasil.filter(...). Pola kedua adalah sebaliknya, yaitu DataFrame hasil transformasi Python yang didaftarkan sebagai view lalu dirangkum dengan SQL. Karena keduanya bertemu di mesin yang sama, perpindahan bolak-balik tidak menimbulkan biaya konversi data, dan topik berikutnya membuktikannya dengan rencana eksekusi.

Satu pertimbangan lain yang menentukan pemilihan SQL adalah keberpihakan pada tim yang meninjau. Kueri yang menjadi keputusan bisnis, misalnya definisi rata-rata atau batas kelulusan, sering lebih cepat ditinjau lintas peran ketika ditulis sebagai SQL, karena bentuknya dekat dengan pernyataan yang dipakai pada basis data relasional sehari-hari. Menempatkan definisi semacam itu dalam SQL yang jelas membuat perubahan definisi terlihat dalam tinjauan kode, alih-alih tersembunyi di dalam rantai fungsi yang panjang. Pilihan ini tidak menutup kemungkinan sisa alurnya tetap ditulis dengan DataFrame, karena keduanya bertemu pada mesin yang sama.

Kesalahan umum#

Kesalahan pertama adalah memanggil spark.sql("SELECT ... FROM nama") ketika view bernama tersebut belum didaftarkan pada sesi aktif. Spark menjawab dengan AnalysisException berbunyi TABLE_OR_VIEW_NOT_FOUND, dan perbaikannya adalah menambahkan createOrReplaceTempView sebelum kueri. Kesalahan ini sering terjadi setelah sesi dibuat ulang, karena view tidak ikut terbawa antar sesi.

Kesalahan kedua adalah salah ketik nama kolom di dalam string SQL. Berbeda dengan API DataFrame yang diperiksa IDE saat diketik, analisis string SQL baru terjadi saat dijalankan, sehingga typo baru ketahuan pada tahap eksekusi. Kebiasaan menjalankan kueri sejak dini dengan .show() pada data kecil membantu menemukannya lebih cepat.

Kesalahan ketiga adalah mengira temp view adalah tabel yang tersimpan. Setelah aplikasi berhenti dengan spark.stop(), nama tersebut hilang dan tidak ada berkas yang tersisa. Apabila data memang perlu dipertahankan antar sesi, jalurnya adalah menulis DataFrame ke format berkas atau tabel sungguhan yang dibahas pada modul berkas dan format.

Ringkasan#

  • Temp view adalah pemetaan nama ke DataFrame yang hidup selama sesi, tanpa penyalinan data dan tanpa berkas yang tersimpan.
  • createOrReplaceTempView mendaftarkan nama yang boleh tertimpa, sedangkan createTempView memunculkan error ketika nama sudah ada.
  • spark.sql() menjalankan pernyataan SQL lengkap terhadap view yang terdaftar dan mengembalikan DataFrame yang bisa dilanjutkan dengan API Python.
  • SQL dan API DataFrame dapat dicampur dalam satu alur karena keduanya bertemu pada mesin yang sama.

Referensi#