Implementasi Data Wrangling dengan PySpark: Transformasi Dataset Skala Besar secara Distributed

Ketika dataset sudah mencapai puluhan gigabyte, Pandas mulai kehabisan akal. Semua data harus dimuat ke dalam memory satu mesin, dan proses itu gagal begitu ukuran data melampaui kapasitas RAM yang tersedia. PySpark hadir sebagai solusi dengan memproses data secara distributed: pekerjaan dipecah menjadi partisi kecil yang dikerjakan paralel oleh banyak worker node.

Apache Spark adalah engine distributed processing yang menjadi standar industri untuk data engineering. Konsep kerjanya sederhana: sebuah driver membagi task ke beberapa worker, setiap worker memproses partisi datanya sendiri, lalu hasilnya digabungkan kembali. Dengan cara ini, kita bisa melakukan wrangling pada dataset yang tidak muat di satu mesin.

Gambar: Ilustrasi arsitektur cluster terdistribusi — master node mengoordinasikan beberapa worker node, konsep yang sama dengan model driver-executor di PySpark — Sumber: Wikimedia Commons — Lisensi: Public Domain

Memahami SparkSession, DataFrame, dan Lazy Evaluation

Entry point untuk semua operasi PySpark adalah SparkSession. Objek ini mengatur konfigurasi cluster, nama aplikasi, dan alokasi resource. Setiap sesi PySpark dimulai dengan membuat SparkSession menggunakan builder pattern.

DataFrame di PySpark berbeda dari DataFrame Pandas. Objek ini terdistribusi di seluruh worker, memiliki schema yang eksplisit, dan bersifat immutable — setiap transformasi menghasilkan DataFrame baru tanpa mengubah yang lama.

Salah satu konsep paling penting adalah lazy evaluation. Transformasi seperti select, filter, dan withColumn tidak langsung dieksekusi. Spark menyusun rencana eksekusi secara internal dan baru menjalankannya ketika sebuah aksi dipanggil, misalnya count(), show(), atau collect(). Mekanisme ini memungkinkan Spark mengoptimalkan seluruh pipeline sebelum diproses.

!pip install pyspark

from pyspark.sql import SparkSession

# Inisialisasi SparkSession
spark = SparkSession.builder \
    .appName("DataWranglingDemo") \
    .master("local[*]") \
    .getOrCreate()

# Membaca dataset sintetis dari file CSV
df = spark.read.csv("transaksi.csv", header=True, inferSchema=True)

# Transformasi TIDAK dieksekusi sampai aksi dipanggil
filtered = df.filter(df["jumlah"] > 1000)

# Aksi: memicu eksekusi dan menampilkan hasil
filtered.show(5)
print(f"Total baris: {filtered.count()}")

Output:

+----+--------+------+------------+
|  id|kategori|jumlah|harga_satuan|
+----+--------+------+------------+
|1001|  Retail|  2500|         300|
|1003|  Retail|  1800|         500|
|1004| Digital|  4500|         200|
|1007|  Retail|  1200|         450|
|1008| Digital|  1500|         300|
+----+--------+------+------------+
only showing top 5 rows
Total baris: 6

Perhatikan bahwa filtered belum berisi data apa pun sampai show() atau count() dipanggil. Ini adalah inti lazy evaluation: kita mendeskripsikan apa yang ingin dilakukan, dan Spark memutuskan kapan dan bagaimana cara paling efisien untuk melakukannya.

Transformasi Data Wrangling dengan DataFrame API

Setelah SparkSession siap, kita bisa melakukan wrangling dengan API yang mirip dengan Pandas namun berjalan secara terdistribusi. Operasi select memilih kolom, filter menyaring baris berdasarkan kondisi, dan withColumn membuat atau mengubah kolom.

Agregasi dilakukan dengan groupBy dan agg. Kombinasi ini setara dengan groupby().agg() di Pandas, tetapi hasilnya dihitung per partisi lalu digabungkan — itulah mengapa operasi ini tetap cepat pada data besar.

Untuk kebutuhan ranking dalam grup, window function adalah alat yang tepat. Fungsi row_number() dan rank() memungkinkan kita memberi nomor urut pada setiap baris berdasarkan partisi dan urutan tertentu.

from pyspark.sql.functions import col, sum, row_number
from pyspark.sql.window import Window

# Seleksi dan filter
transaksi = spark.read.csv("transaksi.csv", header=True, inferSchema=True)
transaksi_retail = transaksi.filter(col("kategori") == "Retail")

# Kolom baru: total nilai transaksi
transaksi = transaksi.withColumn("total_nilai", col("jumlah") * col("harga_satuan"))

# Agregasi per kategori
agregasi = transaksi.groupBy("kategori") \
    .agg(sum("total_nilai").alias("total_penjualan")) \
    .orderBy(col("total_penjualan").desc())

agregasi.show()

# Window function: ranking transaksi terbesar per kategori
window_spec = Window.partitionBy("kategori").orderBy(col("total_nilai").desc())
ranking = transaksi.withColumn("rank", row_number().over(window_spec)) \
    .filter(col("rank") <= 3)

ranking.select("kategori", "id", "total_nilai", "rank").show()

Output:

+--------+---------------+
|kategori|total_penjualan|
+--------+---------------+
|  Retail|        2430000|
| Digital|        2090000|
| Grocery|         189000|
+--------+---------------+

+--------+----+-----------+----+
|kategori|  id|total_nilai|rank|
+--------+----+-----------+----+
| Digital|1004|     900000|   1|
| Digital|1011|     500000|   2|
| Digital|1008|     450000|   3|
| Grocery|1006|      90000|   1|
| Grocery|1012|      84000|   2|
| Grocery|1009|      15000|   3|
|  Retail|1003|     900000|   1|
|  Retail|1001|     750000|   2|
|  Retail|1007|     540000|   3|
+--------+----+-----------+----+

Pipeline di atas berjalan tanpa perlu memuat seluruh data ke memory satu mesin. Setiap transformasi menghasilkan DataFrame baru, dan Spark mengoptimalkan urutan eksekusinya melalui Catalyst Optimizer sebelum aksi dijalankan.

Menangani Missing Values dan Duplikat secara Distributed

Data kotor tidak hanya masalah dataset kecil. Dataset skala besar justru sering membawa lebih banyak null, duplikat, dan inkonsistensi karena berasal dari banyak sumber berbeda. PySpark menyediakan API yang setara dengan Pandas untuk menangani masalah ini.

Deteksi null dilakukan dengan isNull(), isNotNull(), atau dengan menghitung jumlah null per kolom. Untuk handling, dropna() menghapus baris yang mengandung null, sementara fillna() mengisi null dengan nilai tertentu. Kedua operasi ini bisa menerima parameter subset untuk membatasi kolom yang diperiksa.

Duplikat dihapus dengan dropDuplicates(). Perbedaannya dengan Pandas: operasi ini membutuhkan shuffle antar partisi karena baris duplikat bisa berada di partisi yang berbeda. Spark memastikan bahwa deduplikasi tetap benar secara global, bukan hanya per partisi.

# Dataset dengan null dan duplikat
df_kotor = spark.createDataFrame([
    (1, "Andi", 5000000, None),
    (2, "Budi", None, "Retail"),
    (2, "Budi", None, "Retail"),   # duplikat eksak
    (3, "Citra", 6200000, "Digital"),
    (4, "Dewi", None, None),
], ["id", "nama", "gaji", "kategori"])

# Deteksi null per kolom
from pyspark.sql.functions import col, isnull, when, count
df_kotor.select([count(when(isnull(c), c)).alias(c) for c in df_kotor.columns]).show()

# Hapus duplikat berdasarkan subset kolom
df_bersih = df_kotor.dropDuplicates(["id", "nama"])
print(f"Baris sebelum dedup: {df_kotor.count()}, sesudah: {df_bersih.count()}")

# Isi null gaji dengan 0 dan kategori dengan nilai default
df_isi = df_bersih.fillna({"gaji": 0, "kategori": "Unknown"})
df_isi.show()

Output:

+---+----+----+--------+
| id|nama|gaji|kategori|
+---+----+----+--------+
|  0|   0|   3|       2|
+---+----+----+--------+

Baris sebelum dedup: 5, sesudah: 4

+---+-----+-------+--------+
| id| nama|   gaji|kategori|
+---+-----+-------+--------+
|  1| Andi|5000000| Unknown|
|  2| Budi|      0|  Retail|
|  3|Citra|6200000| Digital|
|  4| Dewi|      0| Unknown|
+---+-----+-------+--------+

Strategi handling harus disesuaikan dengan konteks. Menghapus baris dengan dropna() aman jika proporsi null kecil. Mengisi dengan fillna() mempertahankan ukuran dataset, tetapi nilai pengganti harus masuk akal secara domain. Untuk kolom numerik yang skewed, mengisi dengan median per grup lebih akurat daripada mean.

Optimasi Performa dan Best Practices

Performa job PySpark sangat ditentukan oleh bagaimana data dipartisi dan bagaimana operasi dilakukan. repartition() menambah jumlah partisi untuk meningkatkan paralelisme, sementara coalesce() mengurangi partisi tanpa shuffle penuh — cocok untuk mengecilkan hasil sebelum ditulis ke disk.

Join antara tabel besar dan tabel kecil sebaiknya menggunakan broadcast join. Dengan memberikan hint broadcast(), Spark mengirim salinan tabel kecil ke setiap worker sehingga tidak ada shuffle data besar antar node. Penghematan waktu bisa sangat signifikan.

Untuk dataset yang dipakai berulang kali dalam satu job, cache() menyimpan DataFrame di memory sehingga tidak perlu dibaca ulang dari disk setiap kali aksi dipanggil. Format penyimpanan juga berpengaruh: Parquet jauh lebih efisien daripada CSV karena kompresi dan columnar storage.

from pyspark.sql.functions import broadcast

# Cache dataset yang dipakai berulang
transaksi.cache()
transaksi.count()  # memicu load ke memory

# Broadcast join dengan tabel dimensi kecil
dimensi_kategori = spark.createDataFrame([
    ("Retail", "Consumer Goods"),
    ("Digital", "Technology"),
    ("Grocery", "FMCG"),
], ["kategori", "sektor"])

joined = transaksi.join(broadcast(dimensi_kategori), "kategori")
joined.explain()  # lihat execution plan

# Simpan hasil ke Parquet untuk baca yang lebih cepat
joined.write.mode("overwrite").parquet("hasil_wrangling.parquet")

Output:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [kategori#66, id#65, jumlah#67, harga_satuan#68, total_nilai#70, sektor#367]
   +- BroadcastHashJoin [kategori#66], [kategori#366], Inner, BuildRight, false, false
      :- Filter isnotnull(kategori#66)
      :  +- InMemoryTableScan [id#65, kategori#66, jumlah#67, harga_satuan#68, total_nilai#70], [isnotnull(kategori#66)]
      :        +- InMemoryRelation [id#65, kategori#66, jumlah#67, harga_satuan#68, total_nilai#70], StorageLevel(disk, memory, deserialized, 1 replicas)
      :              +- *(1) Project [id#65, kategori#66, jumlah#67, harga_satuan#68, (jumlah#67 * harga_satuan#68) AS total_nilai#70]
      :                 +- FileScan csv [id#65,kategori#66,jumlah#67,harga_satuan#68] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/tmp/blog-sandbox/.../transaksi.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:int,kategori:string,jumlah:int,harga_satuan:int>
      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=532]
         +- Filter isnotnull(kategori#366)
            +- Scan ExistingRDD[kategori#366,sektor#367]

Parquet berhasil ditulis ke hasil_wrangling.parquet

Melihat execution plan dengan explain() adalah kebiasaan yang wajib dibangun. Dari sini kita bisa mendeteksi shuffle yang tidak perlu, melihat apakah broadcast hint diterapkan, dan memastikan prediksi Spark sesuai dengan ekspektasi kita. Spark UI memberikan informasi tambahan tentang waktu setiap stage dan jumlah data yang di-shuffle.

Kemampuan memproses data skala besar secara distributed adalah skill yang semakin dicari di industri, terutama untuk peran data engineer dan data scientist. Di Rumah Coding, program Data Science mengajarkan seluruh pipeline pengolahan data — dari wrangling dengan Pandas, eksplorasi statistik, hingga pemrosesan distributed dengan PySpark untuk dataset yang tidak muat di satu mesin. Bangun portofolio yang siap untuk tantangan data di dunia nyata.