Bagian 4: Melakukan Transformasi

Semua transformasi pada DataFrame bersifat lazy: withColumn, when, dan where hanya menyusun rencana, dan Spark mengeksekusinya ketika aksi seperti show atau count dipanggil. Pada bagian ini kita menerapkan transformasi inti pipeline TokoKita pada 1,6 juta baris pesanan: kolom turunan total, klasifikasi nilai bertingkat, tiga strategi menangani null, lalu perbandingan yang paling mendidik — fungsi built-in melawan UDF Python. Materi pendukung ada pada chapter Spark DataFrames; aspek performanya dibahas pada Spark Performance.

Kolom Turunan dengan withColumn #

Kolom total adalah nilai transaksi per baris: jumlah dikali harga_satuan. Fungsi pada src/transformasi.py:

python
def tambah_total(df):
    """Tambah kolom `total` = jumlah x harga_satuan pada tabel baris pesanan."""
    return df.withColumn(
        "total", F.col("jumlah") * F.col("harga_satuan")
    )

Ekspresi dikompile menjadi kode JVM oleh Catalyst, jadi perkalian ini berjalan secepat kode native tanpa menyentuh Python. Hasilnya pada data nyata (show(8) tanpa orderBy; urutan mengikuti tata letak berkas parquet sehingga stabil antar eksekusi):

Output yang Dihasilkan:

text
== sampel baris pesanan dengan kolom total (withColumn) ==
+-----------+-----+---------+------+------------+------+
|pesanan_id |baris|produk_id|jumlah|harga_satuan|total |
+-----------+-----+---------+------+------------+------+
|ORD-0246168|3    |60       |2     |78000       |156000|
|ORD-0012114|1    |352      |1     |77000       |77000 |
|ORD-0410562|1    |240      |4     |73000       |292000|
|ORD-0528399|2    |4        |2     |91000       |182000|
|ORD-0098062|1    |340      |4     |38000       |152000|
|ORD-0566700|2    |135      |3     |258000      |774000|
|ORD-0250464|1    |2        |1     |87000       |87000 |
|ORD-0130998|1    |156      |5     |55000       |275000|
+-----------+-----+---------+------+------------+------+
only showing top 8 rows

Klasifikasi when Bertingkat #

Klasifikasi membutuhkan logika bercabang: total di atas Rp500.000 disebut besar, di atas Rp200.000 disebut sedang, sisanya kecil. Fungsinya:

python
def klasifikasi_nilai(df, ambang_sedang=200_000, ambang_besar=500_000):
    """Klasifikasi baris pesanan berdasarkan total, contoh when bertingkat."""
    return df.withColumn(
        "kelas_nilai",
        F.when(F.col("total") >= ambang_besar, "besar")
        .when(F.col("total") >= ambang_sedang, "sedang")
        .otherwise("kecil"),
    )

Mari kita telaah baris-baris kode di atas:

Baris 4-6 (urutan when): when dievaluasi berurutan dan berhenti pada kecocokan pertama — sama seperti CASE WHEN pada SQL. Menukar urutan dua ambang mengubah hasilnya.
Baris 2 (parameter ambang): ambang menjadi argumen fungsi, bukan konstanta tersembunyi, sehingga pengujian dapat menggeser ambang tanpa menyunting kode.
Baris 7 (otherwise): wajib ada bila kita menginginkan hasil pasti untuk semua baris; tanpa otherwise, baris yang tidak cocok menghasilkan null.

Hasil dan distribusinya pada seluruh dataset:

Output yang Dihasilkan:

text
== klasifikasi when bertingkat (kelas_nilai) ==
+-----------+-----+------+------------+------+-----------+
|pesanan_id |baris|jumlah|harga_satuan|total |kelas_nilai|
+-----------+-----+------+------------+------+-----------+
|ORD-0246168|3    |2     |78000       |156000|kecil      |
|ORD-0012114|1    |1     |77000       |77000 |kecil      |
|ORD-0410562|1    |4     |73000       |292000|sedang     |
|ORD-0528399|2    |2     |91000       |182000|kecil      |
|ORD-0098062|1    |4     |38000       |152000|kecil      |
|ORD-0566700|2    |3     |258000      |774000|besar      |
|ORD-0250464|1    |1     |87000       |87000 |kecil      |
|ORD-0130998|1    |5     |55000       |275000|sedang     |
+-----------+-----+------+------------+------+-----------+
only showing top 8 rows
distribusi kelas_nilai:
+-----------+------+
|kelas_nilai| count|
+-----------+------+
|      besar|159999|
|      kecil|856665|
|     sedang|583335|
+-----------+------+

Null: Hitung, Buang, atau Isi #

Sistem sumber bisa mengirim baris dengan harga_satuan atau jumlah kosong, dan perkalian apa pun terhadap null menghasilkan null — itulah sebabnya total ikut kosong. Kita memakai contoh empat baris buatan untuk melihat ketiga strategi sekaligus:

Output yang Dihasilkan:

text
== contoh data dengan null (harga_satuan / jumlah kosong) ==
+-----------+-----+---------+------+------------+------+
|pesanan_id |baris|produk_id|jumlah|harga_satuan|total |
+-----------+-----+---------+------+------------+------+
|ORD-9000001|1    |101      |2     |87000       |174000|
|ORD-9000001|2    |205      |1     |NULL        |NULL  |
|ORD-9000002|1    |118      |NULL  |27000       |NULL  |
|ORD-9000002|2    |301      |1     |252000      |252000|
+-----------+-----+---------+------+------------+------+

ringkasan null: {'jumlah': 1, 'harga_satuan': 1, 'total': 2}
bersihkan null : {'null_sebelum': {'jumlah': 1, 'harga_satuan': 1}, 'baris_tanpa_null': 2, 'baris_terisi': 4}
setelah fillna(0), total baris 1 (jumlah 2 x 87000 = 174000):
+-----------+-----+---------+------+------------+------+
|pesanan_id |baris|produk_id|jumlah|harga_satuan|total |
+-----------+-----+---------+------+------------+------+
|ORD-9000001|1    |101      |2     |87000       |174000|
|ORD-9000001|2    |205      |1     |0           |0     |
|ORD-9000002|1    |118      |0     |27000       |0     |
|ORD-9000002|2    |301      |1     |252000      |252000|
+-----------+-----+---------+------+------------+------+

Tiga strategi terbaca langsung dari keluaran tersebut. ringkasan_null menghitung null per kolom (2 dari 4 baris). dropna membuang dua baris yang tidak lengkap (baris_tanpa_null: 2). fillna(0) mengisi nilai nol sehingga total baris tetap empat — pilihan ini mengubah makna bisnis: baris tanpa harga menjadi transaksi bernilai nol, jadi keputusan memilih antara membuang dan mengisi adalah keputusan bisnis, bukan sekadar teknik.

UDF Python vs Fungsi Built-in #

Kita menulis total versi UDF: fungsi Python biasa berlabel @F.udf:

python
    @F.udf(returnType=IntegerType())
    def jumlah_x_harga(jumlah, harga_satuan):
        return None if jumlah is None or harga_satuan is None else jumlah * harga_satuan

Kedua varian lalu diadu pada aksi SUM(total) yang sama:

Output yang Dihasilkan:

text
== UDF vs built-in pada 400 ribu baris (SUM(total), rata-rata 2x) ==
  hasil: {'builtin_s': 0.738, 'udf_s': 1.918, 'baris': 399999}
  UDF 2.6x lebih lambat dari built-in pada dataset ini

Angka tersebut memerlukan keterbukaan soal skala. Pada percobaan pertama, perbandingan dijalankan terhadap seluruh 1,6 juta baris dan worker Python berhenti di tengah pengukuran — mesin uji dua CPU dengan 3,9 GB RAM tidak sanggup memproses UDF serialisasi per baris pada volume itu. Perbandingan kemudian diulang pada potongan 400 ribu baris pertama (pesanan nomor 1 sampai 200.000) dan stabil: UDF 2,6 kali lebih lambat dari built-in. Penyebabnya bukan fungsi perkaliannya, melainkan biaya serialisasi: setiap baris harus dinaikkan ke Python lalu diturunkan kembali ke JVM, sedangkan built-in dieksekusi sepenuhnya di dalam JVM.

Transformasi selesai dan total siap dipakai. Kolom ini menjadi bahan baku bagian berikutnya, tempat baris pesanan bertemu header pesanan dan dimensi produk: join, agregasi bulanan, dan pilihan strategi join oleh optimizer — tugas Bagian 5.