Belajar Apache Spark - Joins, Aggregations, & Window Functions

#Pendahuluan

Setelah menguasai DataFrame dan SQL dasar, episode 7 ini masuk ke operasi analitik yang paling sering dipakai di dunia nyata: join, agregasi, dan window function. Ketiganya adalah bahasa sehari-hari data analyst dan data engineer.

Di episode ini kita membahas semua tipe join yang tersedia, agregasi groupBy dan variasi lanjutannya, window function untuk analisis berbasis urutan waktu, serta implikasi performa — karena join dan shuffle adalah dua penyebab paling umum query melambat.

#Tipe-Tipe Join di Spark

#Inner Join dan Family of Joins

Spark mendukung seluruh tipe join standar SQL. Contoh paling jelas memakai PySpark:

Inner join dua DataFrame

pelanggan = spark.createDataFrame([(1, "budi"), (2, "sari")], ["id", "nama"])
pesanan = spark.createDataFrame([(1, "p1"), (1, "p2"), (3, "p3")], ["id", "produk"])
 
inner = pelanggan.join(pesanan, "id", "inner")
inner.show()

pelanggan.join(pesanan, "id", "inner") hanya mengembalikan baris yang id-nya cocok di kedua sisi. Tipe join yang tersedia:

• inner: hanya baris yang cocok.
• left outer: semua baris kiri, sisanya null di kanan.
• right outer: semua baris kanan, sisanya null di kiri.
• full outer: gabungan semua baris kedua sisi.
• left semi: baris kiri yang punya pasangan — kolom kanan tidak ikut.
• left anti: baris kiri yang tidak punya pasangan — berguna untuk deteksi data hilang.

left semi → "data yang punya pasangan" (filter)
left anti → "data yang tidak punya pasangan" (data quality check)

left_anti misalnya sangat praktis untuk menemukan pesanan yang id pelanggannya tidak ada di tabel pelanggan.

#Cross Join dan Hati-Hati

cross join menghasilkan perkalian semua baris kedua sisi. Karena hasilnya bisa meledak secara eksponensial, Spark memerlukan konfirmasi eksplisit sebelum menjalankannya.

#GroupBy dan Agregasi

#Agregasi Dasar dan Advanced Grouping

groupBy menggabungkan baris berdasarkan kunci lalu menerapkan fungsi agregasi:

Agregasi dengan groupBy

from pyspark.sql import functions as F
 
penjualan = spark.read.option("header", True).csv("data/penjualan.csv")
hasil = penjualan \
    .groupBy("kota", "kategori") \
    .agg(
        F.sum("jumlah").alias("total"),
        F.avg("harga").alias("harga_rata"),
        F.countDistinct("produk").alias("produk_unik"),
    )
hasil.show()

F.sum, F.avg, dan F.countDistinct adalah fungsi agregasi bawaan. groupBy bisa menerima beberapa kolom untuk granularitas lebih dalam — analisis kota per kategori, misalnya.

#Rollup, Cube, dan Grouping Sets

Untuk analisis multidimensi, Spark menyediakan tiga variasi:

• rollup: agregasi berjenjang dari total menuju kelompok paling detail.
• cube: semua kombinasi kunci, termasuk subtotal.
• grouping sets: kumpulan kelompok yang dipilih eksplisit.

rollup(a, b) → (a,b), (a), (keseluruhan)
cube(a, b)   → (a,b), (a), (b), (keseluruhan)

cube menghasilkan lebih banyak baris karena mencakup semua kombinasi. Ini berguna untuk laporan yang perlu melihat data dari berbagai sudut tanpa menulis beberapa query terpisah.

#Rolling Aggregations dengan Window

Untuk agregasi yang bergerak pada jendela urutan (misalnya rata-rata 7 hari terakhir), gabungkan groupBy dengan window function di bagian berikut.

#Window Functions

#Window: Agregasi Tanpa Menggabungkan Baris

Berbeda dengan groupBy yang menciutkan baris, window function menghitung nilai agregasi namun mempertahankan setiap baris. Definisikan window dengan partitionBy dan orderBy:

Row number per partisi

from pyspark.sql.window import Window
 
w = Window.partitionBy("kota").orderBy(F.col("total").desc())
 
hasil = penjualan.withColumn("peringkat", F.row_number().over(w))
hasil.filter(F.col("peringkat") == 1).show()

F.row_number().over(w) memberi nomor urut di dalam setiap partisi kota berdasarkan total terbesar. Hasilnya: top 1 penjualan per kota — tanpa kehilangan kolom lain.

#Fungsi Window Penting

Fungsi yang paling sering dipakai:

• row_number(): nomor urut per partisi.
• rank() dan dense_rank(): peringkat dengan penanganan nilai sama.
• lag() dan lead(): mengambil nilai baris sebelumnya dan berikutnya.
• sum(), avg() dengan window: agregasi berjalan (rolling).

Perubahan nilai antar periode

w2 = Window.partitionBy("kota").orderBy("bulan")
 
hasil2 = penjualan.withColumn("bulan_lalu", F.lag("total").over(w2)) \
                  .withColumn("delta", F.col("total") - F.col("bulan_lalu"))

F.lag("total").over(w2) mengambil nilai total dari bulan sebelumnya dalam kota yang sama — pola dasar analisis time-series seperti perhitungan growth.

#Time-Based Analytics dengan Window Grouping

Untuk agregasi berdasarkan rentang waktu (misalnya per jam), gunakan fungsi F.window:

Agregasi per jendela waktu

from pyspark.sql import functions as F
 
event = spark.read.json("data/event.json")
per_jam = event \
    .groupBy(F.window("waktu", "1 hour"), "produk") \
    .count()
per_jam.show()

F.window("waktu", "1 hour") mengelompokkan waktu ke jendela per jam — pola yang akan kembali muncul di Structured Streaming pada episode 10.

#Implikasi Performa Join dan Shuffle

#Mengapa Join Mahal

Sebagian besar join membutuhkan shuffle: data dikelompokkan ulang ke partisi berdasarkan kunci join, dikirim antar executor melalui jaringan. Semakin besar data, semakin besar biayanya. Ada beberapa strategi untuk mengurangi biaya ini:

• Broadcast join: jika satu sisi kecil, kirim salinan ke semua executor sehingga tidak perlu shuffle sama sekali.
• Sort-merge join: strategi default untuk data besar yang memakai sort dan merge.
• Bucketting: menyiapkan data dengan bucketing di kunci join agar join menjadi lokal.

broadcast (sisi kecil) → bucket join (disiapkan) → sort-merge (default)

Detail strategi dan kapan memakainya akan dibahas tuntas di episode 9. Yang perlu kalian ingat sekarang: setiap join dan groupBy yang memicu shuffle adalah kandidat utama optimasi.

#Penutup

Episode 7 membekali kalian tiga operasi analitik inti: join dengan delapan tipe dari inner sampai anti, agregasi groupBy dengan rollup dan cube, serta window function untuk analisis berurutan dan berbasis waktu. Kalian juga memahami mengapa join dan shuffle adalah sumber biaya utama.

Inti yang harus dibawa pulang:

• Spark mendukung inner, outer, semi, anti, dan cross join.
• left_anti adalah alat deteksi data yang hilang di tabel referensi.
• rollup dan cube menambah dimensi subtotal pada agregasi.
• Window function menghitung agregasi tanpa menciutkan baris.
• Setiap join besar berpotensi shuffle besar — antisipasi sejak awal.

Di episode 8 selanjutnya kita akan membahas data sources dan storage — menghubungkan Spark ke HDFS, S3, JDBC, dan file system, membaca serta menulis Parquet, Avro, ORC, JSON, dan CSV, serta strategi partitioning, bucketing, file layout, dan schema evolution.