KAL
Menu

Pengalaman singkat PySpark

Menjalankan PySpark pada data sampel, yakni API DataFrame dan operasi dasarnya.

Daftar isi

PySpark adalah pustaka Python resmi untuk Apache Spark dan menjadi cara paling mudah mencoba Spark tanpa menulis kode Scala atau Java. Topik ini memandu pengalaman pertama menjalankan PySpark pada data sampel, meliputi penyiapan SparkSession, pembuatan DataFrame kecil, serta operasi filter dan groupBy. Seluruh contoh dijalankan pada mode local, yaitu tanpa cluster, sehingga eksekusinya berlangsung di satu mesin.

Mode local dan SparkSession#

SparkSession adalah titik masuk setiap aplikasi yang memakai API DataFrame, dan setiap skrip PySpark dimulai dari objek ini. Pada mode local, driver beserta seluruh thread pekerjanya berjalan di dalam satu proses JVM (Java Virtual Machine), sehingga tidak ada jaringan dan tidak ada node lain yang perlu disiapkan. Parameter local[1] menetapkan satu thread pekerja, dan angkanya dapat dinaikkan, misalnya local[4], agar jumlah thread yang membagi pekerjaan bertambah.

PySpark terpasang melalui pip tanpa instalasi Spark terpisah, dan versi yang terpasang di lingkungan jalur ini dapat diperiksa secara langsung.

bash 
# memeriksa versi PySpark yang terpasang
python3 -c "import pyspark; print(pyspark.__version__)"
text Keluaran 
4.3.0

Mengimpor pustaka pyspark saja tidak memulai JVM, sehingga pemeriksaan versi selesai dalam sepersekian detik. JVM baru dimulai ketika SparkSession dibuat, dan proses tersebut memakan waktu belasan detik sebelum baris kode pertama benar-benar berjalan. Waktu pemanasan seperti ini menjadi salah satu ciri pekerjaan Spark yang tidak ditemui pada alat proses tunggal seperti pandas atau DuckDB.

Skrip pertama pada data sampel#

Skrip berikut merangkum alur paling dasar PySpark, yaitu memulai SparkSession pada mode local, membuat DataFrame kecil dari daftar tuple, lalu menampilkan isinya. Alur dilanjutkan dengan operasi filter dan agregasi groupBy, kemudian sesinya dihentikan di akhir skrip.

python pyspark-kecil.py 
# tidak-dijalankan
from pyspark.sql import SparkSession

# memulai SparkSession pada mode local, yaitu seluruh eksekusi berlangsung di satu JVM
spark = SparkSession.builder.master("local[1]").appName("PysparkKecil").getOrCreate()

# membuat DataFrame kecil dari daftar tuple
data = [
    ("Ali", "matematika", 88),
    ("Ali", "fisika", 91),
    ("Budi", "matematika", 75),
    ("Citra", "fisika", 83),
    ("Citra", "biologi", 79),
]
df = spark.createDataFrame(data, ["nama", "mata_pelajaran", "nilai"])

# show() adalah action, yaitu pemicu eksekusi yang menampilkan isi DataFrame
df.show()

# filter dan select termasuk transformation, sehingga belum ada pekerjaan yang berjalan
lulus = df.filter(df.nilai >= 80).select("nama", "mata_pelajaran")
lulus.show()

# groupBy memulai agregasi, dan count() beserta show() menampilkannya
df.groupBy("nama").count().orderBy("nama").show()

# menghentikan SparkSession agar sumber dayanya dibebaskan
spark.stop()

Skrip ini tidak dieksekusi utuh di dalam artikel karena keluaran mentahnya bercampur dengan log JVM, yaitu baris konfigurasi, cap waktu, dan penghitung kemajuan eksekusi yang membuat keluaran tidak bersih dibaca. Keluaran tabelnya sendiri berformat tetap, sehingga blok berikut menampilkan format tersebut sebagaimana didokumentasikan pada panduan resmi Spark.

text Keluaran yang diharapkan 
+-----+--------------+-----+
| nama|mata_pelajaran|nilai|
+-----+--------------+-----+
|  Ali|    matematika|   88|
|  Ali|        fisika|   91|
| Budi|    matematika|   75|
|Citra|        fisika|   83|
|Citra|       biologi|   79|
+-----+--------------+-----+

Format tabel dengan garis +---+ adalah format show() bawaan Spark, dengan setiap kolom diratakan mengikuti nilai terpanjangnya. Method createDataFrame menerima daftar tuple beserta daftar nama kolom, dan Spark menetapkan tipe datanya secara otomatis dari nilai yang diberikan, yaitu string untuk nama dan mata pelajaran serta bilangan bulat untuk nilai. Data sampel sekecil ini sebenarnya tidak membutuhkan Spark, dan tujuannya hanya membiasakan bentuk API-nya sebelum data yang sesungguhnya besar.

Filter dan groupBy#

Operasi pada DataFrame mengikuti pola yang dekat dengan SQL, dan dua operasi yang paling sering dipakai adalah filter untuk menyaring baris serta groupBy untuk agregasi. Bagian skrip berikut menyaring baris dengan nilai minimal 80, lalu memilih dua kolomnya.

python Contoh: filter dan select pada data sampel 
# tidak-dijalankan
lulus = df.filter(df.nilai >= 80).select("nama", "mata_pelajaran")
lulus.show()
text Keluaran yang diharapkan 
+-----+--------------+
| nama|mata_pelajaran|
+-----+--------------+
|  Ali|    matematika|
|  Ali|        fisika|
|Citra|        fisika|
+-----+--------------+

Baris yang tampil hanya milik Ali dan Citra, yaitu dua nama dengan nilai minimal 80 sesuai syarat filternya. Pada agregasi berikut, groupBy mengelompokkan baris per nama, count menghitung banyaknya baris pada tiap kelompok, dan orderBy mengurutkan hasilnya berdasarkan nama.

python Contoh: agregasi groupBy dan count 
# tidak-dijalankan
df.groupBy("nama").count().orderBy("nama").show()
text Keluaran yang diharapkan 
+-----+-----+
| nama|count|
+-----+-----+
|  Ali|    2|
| Budi|    1|
|Citra|    2|
+-----+-----+

Ali dan Citra masing-masing mengambil dua mata pelajaran, sedangkan Budi satu, dan hasilnya konsisten dengan data sampelnya. Ketiga keluaran pada topik ini mengikuti format show() yang didokumentasikan secara resmi, dan pencantumannya sebagai keluaran yang diharapkan menghindari campuran log JVM yang telah dijelaskan. Baris terakhir skrip memanggil spark.stop() agar proses JVM dan sumber daya yang dipegangnya dibebaskan pada akhir pekerjaan.

Menjalankan skrip dengan spark-submit#

Cara standar menjalankan berkas PySpark adalah perintah spark-submit, yakni peluncur resmi Spark yang juga dipakai untuk mengirim aplikasi ke cluster.

bash 
# tidak-dijalankan
# memasang PySpark bila belum terpasang
pip install pyspark
# menjalankan skrip pada mode local tanpa cluster
spark-submit pyspark-kecil.py

Bila berkas dijalankan lewat spark-submit tanpa parameter tambahan, mode local yang dinyatakan di dalam kode tetap dipakai. Pengiriman ke cluster sungguhan menetapkan targetnya melalui opsi --master pada spark-submit, dan baris .master() pada kode perlu dilepas agar tidak bertentangan. Kode aplikasinya sendiri tidak berubah, dan perbedaannya hanya berada pada cara peluncurannya.

Kesalahan umum#

Kesalahan berikut paling sering ditemui pada percobaan PySpark pertama, dan setiap butir menyertakan perbaikannya.

  • Mencampur objek pandas dan PySpark pada satu operasi. Keduanya memiliki nama kelas yang mirip tetapi API-nya tidak saling menggantikan, sehingga operasi campuran berakhir dengan error atribut. Perbaikannya adalah menjaga satu jenis DataFrame pada satu tahap dan mengonversi secara eksplisit bila memang diperlukan.
  • Memanggil show() berulang kali pada DataFrame yang sama tanpa cache. Setiap show() adalah aksi baru yang mengulang seluruh rencana dari awal. Perbaikannya adalah memanggil cache() pada DataFrame yang dipakai lebih dari sekali.
  • Lupa memanggil spark.stop(). Proses JVM tetap hidup dan memegang memori sampai skripnya benar-benar berakhir. Perbaikannya adalah menutup sesi pada akhir skrip sebagaimana contoh.

Ringkasan#

  • SparkSession adalah titik masuk aplikasi DataFrame, dan mode local menjalankan driver beserta thread pekerjanya di satu proses JVM tanpa cluster.
  • createDataFrame membentuk DataFrame dari data Python, sedangkan show() menampilkannya dalam format tabel berpembatas +---+ yang tetap.
  • Filter dan select adalah transformasi, sedangkan show() dan count() adalah aksi yang memicu eksekusi rencana.
  • spark-submit adalah peluncur standar untuk berkas PySpark, baik pada mode local maupun pada cluster melalui opsi --master.

Referensi#