Bagian 5: Melakukan Join dan Agregasi

Nilai transaksi per baris baru berguna setelah bertemu konteksnya: status dari header pesanan, nama produk dari dimensi. Pada bagian ini kita menyusun gabungan tiga tabel itu, lalu menghadapi pertanyaan yang membedakan pengguna Spark pemula dari yang berpengalaman: join mana yang dipilih optimizer, dan berapa harganya? Kita mengukur sendiri BroadcastHashJoin melawan SortMergeJoin dengan membalik satu konfigurasi, membaca jenis join langsung dari physical plan, lalu menutup dengan dua agregasi bisnis: metrik bulanan 24 bulan dan sepuluh produk terlaris. Materi pendukung ada pada chapter Spark Joins dan Shuffle dan Skew.

Potongan Desember 2025 untuk Pengukuran #

Pengukuran yang adil membutuhkan potongan data yang bisa diverifikasi. Dari rumus pembangkit Bagian 1 — pesanan ke-n jatuh pada 2024-01-01 ditambah (n-1) % 731 hari — pesanan Desember 2025 adalah mereka yang (nomor - 1) % 731 bernilai 700 sampai 730. Kita mengambil dua potongan secara berbeda: header lewat filter kolom partition bulan, baris lewat rumus nomor, lalu mencocokkan jumlahnya:

Output yang Dihasilkan:

text
== potongan Desember 2025 (rumus pembangkit: (nomor-1) % 731 == 700..730) ==
  pesanan Des 2025: 33,914 baris
  baris  Des 2025: 67,827 baris

Angka 33.914 persis dua kali lipat kurang sedikit dari 67.827 — masuk akal karena setiap pesanan berisi satu sampai tiga baris. Kedua potongan ini menjadi sisi kiri dan kanan pengukuran join berikut.

Broadcast vs Sort-Merge #

Fungsi ukur_join pada src/joins.py menjalankan join yang sama dua kali: varian pertama dengan ambang broadcast 10 MB (bawaan), varian kedua dengan spark.sql.autoBroadcastJoinThreshold=-1 sehingga broadcast dimatikan dan planner terpaksa memilih sort-merge. Konfigurasi dikembalikan ke semula di blok finally, dan jenis join dibaca dari plan sebagai bukti — bukan dikira-kira:

Output yang Dihasilkan:

text
== ukur_join: broadcast vs sort-merge (kunci pesanan_id, 2x) ==
  broadcast   jenis=BroadcastHashJoin  durasi=0.978 s baris=67,827
  sort_merge  jenis=SortMergeJoin      durasi=1.268 s baris=67,827

Kedua varian menghasilkan 67.827 baris yang sama, jadi selisih 0,978 lawan 1,268 detik (sekitar 23 persen) murni biaya algoritma. Mekanismenya berbeda total: BroadcastHashJoin mengirim seluruh sisi kecil ke semua task sehingga join terjadi tanpa perpindahan data (tanpa shuffle), sedangkan SortMergeJoin mengacak kedua sisi dengan kunci yang sama (Exchange hashpartitioning), mengurutkannya, lalu menggabungkan — biaya shuffle inilah yang dibahas pada chapter Shuffle dan Skew. Pembacaan jenis join dari plan dilakukan fungsi kecil ini:

python
def jenis_join(df):
    """Baca jenis physical join dari query plan (BroadcastHashJoin/SortMergeJoin)."""
    teks = df._jdf.queryExecution().executedPlan().toString()
    for kandidat in ("BroadcastHashJoin", "SortMergeJoin", "BroadcastNestedLoopJoin"):
        if kandidat in teks:
            return kandidat
    return "tidak terdeteksi"

Mari kita telaah baris-baris kode di atas:

Baris 3 (executedPlan().toString()): queryExecution() adalah query execution JVM pada DataFrame (atribut _jdf); executedPlan adalah physical plan siap eksekusi, dan teksnya memuat nama operator seperti BroadcastHashJoin.
Baris 4-6 (daftar kandidat): pencocokan dilakukan terhadap tiga jenis join yang paling sering muncul; urutan pencarian tidak berpengaruh karena satu plan hanya memuat satu jenis untuk satu join.
Baris 7 (nilai balik): tidak terdeteksi menjadi penanda jujur bahwa fungsi ini alat diagnostik, bukan jaminan.

Pada data penuh, jenis join tetap terbaca tanpa mengeksekusi satu baris pun — cukup menyusun rencananya:

Output yang Dihasilkan:

text
== jenis join pada data penuh (dibaca dari plan) ==
  baris x pesanan (800 ribu): BroadcastHashJoin
  baris x pesanan, ambang -1: SortMergeJoin

Gabungan Tiga Tabel #

Rantai lengkapnya: tambah_total menghitung nilai baris, gabung_baris_pesanan membawa status dan bulan dari header (kunci pesanan_id, string), filter_status membuang yang tidak selesai, dan gabung_produk membawa nama serta kategori dari dimensi (kunci produk_id). Delapan baris pertamanya:

Output yang Dihasilkan:

text
== gabungan baris x pesanan x produk (contoh tampilan) ==
+-----------+-----+-------------------------+------+------------+------+-------+
|pesanan_id |baris|nama_produk              |jumlah|harga_satuan|total |status |
+-----------+-----+-------------------------+------+------------+------+-------+
|ORD-0246168|3    |Mustika Spesial kecil    |2     |78000       |156000|selesai|
|ORD-0012114|1    |Senja Spesial besar      |1     |77000       |77000 |selesai|
|ORD-0410562|1    |Mustika Spesial kecil    |4     |73000       |292000|selesai|
|ORD-0098062|1    |Senja Ekonomis besar     |4     |38000       |152000|selesai|
|ORD-0250464|1    |Gayo Gold Signature kecil|1     |87000       |87000 |selesai|
|ORD-0768211|1    |Rumah Kopi Ekonomis besar|3     |61000       |183000|selesai|
|ORD-0775777|1    |Awan Signature besar     |4     |28000       |112000|selesai|
|ORD-0218948|1    |Gayo Gold Premium kecil  |5     |65000       |325000|selesai|
+-----------+-----+-------------------------+------+------------+------+-------+
only showing top 8 rows

Metrik Bulanan dan Produk Terlaris #

Agregasi bulanan menghitung empat metrik per bulan: GMV, jumlah pesanan, jumlah baris, dan rata-rata nilai baris. countDistinct membuat Spark bekerja dua tingkat — agregasi parsial per pasangan bulan-pesanan, lalu penggabungan per bulan; kita bedah rencananya pada Bagian 7:

Output yang Dihasilkan:

text
== agregasi bulanan seluruh dataset (24 bulan) ==
+-------+-----------+--------------+------------+---------------+
|bulan  |gmv        |jumlah_pesanan|jumlah_baris|rata_rata_baris|
+-------+-----------+--------------+------------+---------------+
|2024-01|15411002000|28854         |57710       |267042.1       |
|2024-02|14419678000|26991         |53980       |267130.0       |
|2024-03|15416477000|28854         |57710       |267137.0       |
|2024-04|14919078000|27922         |55842       |267165.9       |
|2024-05|15408471000|28854         |57709       |267002.9       |
|2024-06|14915330000|27923         |55846       |267079.6       |
|2024-07|15414284000|28854         |57707       |267112.9       |
|2024-08|15415362000|28853         |57708       |267126.9       |
|2024-09|14913343000|27923         |55844       |267053.6       |
|2024-10|15399552000|28835         |57670       |267028.8       |
|2024-11|14904201000|27898         |55797       |267114.7       |
|2024-12|15403304000|28826         |57650       |267186.5       |
|2025-01|15399526000|28826         |57655       |267097.8       |
|2025-02|13905792000|26038         |52074       |267039.1       |
|2025-03|15401085000|28827         |57657       |267115.6       |
|2025-04|14892715000|27897         |55790       |266942.4       |
|2025-05|15404216000|28826         |57655       |267179.2       |
|2025-06|14905633000|27897         |55792       |267164.3       |
|2025-07|15394023000|28826         |57651       |267020.9       |
|2025-08|15394979000|28827         |57657       |267009.7       |
|2025-09|14900646000|27896         |55789       |267089.3       |
|2025-10|15404878000|28828         |57658       |267176.8       |
|2025-11|14902410000|27896         |55792       |267106.6       |
|2025-12|15397383000|28829         |57656       |267056.0       |
+-------+-----------+--------------+------------+---------------+

  durasi agregasi+show: 5.40 s

GMV per bulan stabil di sekitar Rp15 miliar karena rumus pembangkit menyebarkan pesanan merata; Februari lebih rendah (28 hari). Stabilitas seperti ini justru memudahkan kita mendeteksi anomali pada data nyata. Terakhir, sepuluh produk terlaris berdasarkan jumlah item:

Output yang Dihasilkan:

text
== 10 produk terlaris (jumlah item) ==
+------------------------+---------------+-----------+----------+
|nama_produk             |kategori       |jumlah_item|gmv       |
+------------------------+---------------+-----------+----------+
|Karta Ekonomis sedang   |alat seduh     |17337      |4507620000|
|Delima Signature kecil  |kopi bubuk     |17337      |1473645000|
|Lentera Signature kecil |perlengkapan   |17337      |1473645000|
|Pagi Hari Premium kecil |kopi bubuk     |17337      |1473645000|
|Gayo Gold Premium sedang|kopi saset     |17337      |1213590000|
|Gayo Gold Premium kecil |sirup          |17337      |1126905000|
|Karta Ekonomis kecil    |sirup          |17337      |1126905000|
|Pagi Hari Premium besar |susu dan krimer|17337      |780165000 |
|Rumah Kopi Spesial besar|susu dan krimer|17337      |780165000 |
|Karta Ekonomis besar    |cemilan        |17337      |520110000 |
+------------------------+---------------+-----------+----------+

Seluruh jumlah_item seri di angka 17.337 — rumus (nomor * 31 + baris * 17) % 400 memang membagi produk secara merata — sehingga urutan akhir ditentukan orderBy(F.desc("jumlah_item"), F.desc("gmv")): dua kunci, tie-break GMV. Konfigurasi ambang broadcast yang kita balik-balik tadi terdokumentasi di https://spark.apache.org/docs/latest/configuration.html; panduan lengkap SQL dan optimizer ada di https://spark.apache.org/docs/latest/ .

Join dan agregasi selesai. Namun sebelum menulis hasil, satu pertanyaan teknis belum terjawab: berapa berkas yang harus dihasilkan, dan bagaimana mengaturnya? Itu pekerjaan Bagian 6.