Spark Joins

Hampir tidak ada analitik yang berdiri di satu tabel: omzet per pesanan butuh nama pelanggan, laporan produk butuh kategori, dan rekomendasi butuh riwayat. Join adalah operasi yang menyatukan semua itu — sekaligus operasi termahal di Spark karena kunci yang cocok sering berada di partition yang berbeda, sehingga data harus berpindah lewat shuffle. Pada chapter ini kita menggabungkan pesanan TokoKita dengan data pelanggan memakai tiga tipe join utama, membuktikan bahaya fan-out yang melipatgandakan baris secara diam-diam, lalu menekan biaya shuffle dengan broadcast join. Semua bukti kita baca langsung dari hasil eksekusi dan rencana eksekusi nyata, bukan dari teori.

Skrip Lengkap dan Outputnya #

Skrip berikut memakai dua tabel kecil — pelanggan berisi lima orang dan pesanan berisi delapan pesanan — dengan dua jebakan yang sengaja ditanam: pesanan 106 memakai pelanggan_id 99 yang tidak ada di tabel pelanggan, dan pelanggan Agus tidak memiliki satu pesanan pun.

python
"""Spark joins: inner/left/full outer, fan-out, dan broadcast join."""
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, col, sum as f_sum

spark = (
    SparkSession.builder
    .master("local[1]")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")

pelanggan = spark.createDataFrame(
    [
        (1, "Budi", "Bandung"),
        (2, "Siti", "Jakarta"),
        (3, "Dewi", "Surabaya"),
        (4, "Rina", "Bandung"),
        (5, "Agus", "Medan"),
    ],
    "pelanggan_id INT, nama STRING, kota STRING",
)
pesanan = spark.createDataFrame(
    [
        (101, 1, "Kemeja Batik", 2),
        (102, 2, "Kain Songket", 1),
        (103, 1, "Rok Tenun", 3),
        (104, 3, "Blouse Katun", 1),
        (105, 2, "Kemeja Batik", 2),
        (106, 99, "Kain Songket", 1),
        (107, 4, "Rok Tenun", 1),
        (108, 3, "Kain Songket", 1),
    ],
    "pesanan_id INT, pelanggan_id INT, produk STRING, qty INT",
)

print("== 1. Inner join: hanya pesanan dengan pelanggan yang cocok ==")
pesanan.join(pelanggan, "pelanggan_id", "inner").orderBy("pesanan_id").show()

print("== 2. Left join: semua pesanan dipertahankan ==")
pesanan.join(pelanggan, "pelanggan_id", "left").orderBy("pesanan_id").show()

print("== 3. Full outer join: kedua sisi dipertahankan ==")
pesanan.join(pelanggan, "pelanggan_id", "full_outer").orderBy("pesanan_id").show()

print("== 4. Fan-out: satu pesanan menempel ke banyak tag ==")
tag = spark.createDataFrame(
    [
        ("Kemeja Batik", "pakaian"), ("Kemeja Batik", "formal"),
        ("Kain Songket", "pakaian"), ("Kain Songket", "souvenir"),
        ("Rok Tenun", "pakaian"), ("Rok Tenun", "formal"),
        ("Blouse Katun", "pakaian"), ("Blouse Katun", "kasual"),
    ],
    "produk STRING, tag STRING",
)
pesanan.join(tag, "produk").filter(col("pesanan_id") == 101).show()

total_qty = pesanan.groupBy().sum("qty").collect()[0][0]
total_qty_join = pesanan.join(tag, "produk").groupBy().sum("qty").collect()[0][0]
print(f"Total qty sebelum join : {total_qty}")
print(f"Total qty setelah join : {total_qty_join}")

print("== 5. Broadcast join: rencana eksekusi dengan dan tanpa hint ==")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
print("Tanpa hint (broadcast dimatikan):")
pesanan.join(pelanggan, "pelanggan_id").explain()
print("Dengan broadcast(pelanggan):")
pesanan.join(broadcast(pelanggan), "pelanggan_id").explain()

Output yang Dihasilkan:

text
== 1. Inner join: hanya pesanan dengan pelanggan yang cocok ==
+------------+----------+------------+---+----+--------+
|pelanggan_id|pesanan_id|      produk|qty|nama|    kota|
+------------+----------+------------+---+----+--------+
|           1|       101|Kemeja Batik|  2|Budi| Bandung|
|           2|       102|Kain Songket|  1|Siti| Jakarta|
|           1|       103|   Rok Tenun|  3|Budi| Bandung|
|           3|       104|Blouse Katun|  1|Dewi|Surabaya|
|           2|       105|Kemeja Batik|  2|Siti| Jakarta|
|           4|       107|   Rok Tenun|  1|Rina| Bandung|
|           3|       108|Kain Songket|  1|Dewi|Surabaya|
+------------+----------+------------+---+----+--------+

== 2. Left join: semua pesanan dipertahankan ==
+------------+----------+------------+---+----+--------+
|pelanggan_id|pesanan_id|      produk|qty|nama|    kota|
+------------+----------+------------+---+----+--------+
|           1|       101|Kemeja Batik|  2|Budi| Bandung|
|           2|       102|Kain Songket|  1|Siti| Jakarta|
|           1|       103|   Rok Tenun|  3|Budi| Bandung|
|           3|       104|Blouse Katun|  1|Dewi|Surabaya|
|           2|       105|Kemeja Batik|  2|Siti| Jakarta|
|          99|       106|Kain Songket|  1|NULL|    NULL|
|           4|       107|   Rok Tenun|  1|Rina| Bandung|
|           3|       108|Kain Songket|  1|Dewi|Surabaya|
+------------+----------+------------+---+----+--------+

== 3. Full outer join: kedua sisi dipertahankan ==
+------------+----------+------------+----+----+--------+
|pelanggan_id|pesanan_id|      produk| qty|nama|    kota|
+------------+----------+------------+----+----+--------+
|           5|      NULL|        NULL|NULL|Agus|   Medan|
|           1|       101|Kemeja Batik|   2|Budi| Bandung|
|           2|       102|Kain Songket|   1|Siti| Jakarta|
|           1|       103|   Rok Tenun|   3|Budi| Bandung|
|           3|       104|Blouse Katun|   1|Dewi|Surabaya|
|           2|       105|Kemeja Batik|   2|Siti| Jakarta|
|          99|       106|Kain Songket|   1|NULL|    NULL|
|           4|       107|   Rok Tenun|   1|Rina| Bandung|
|           3|       108|Kain Songket|   1|Dewi|Surabaya|
+------------+----------+------------+----+----+--------+

== 4. Fan-out: satu pesanan menempel ke banyak tag ==
+------------+----------+------------+---+-------+
|      produk|pesanan_id|pelanggan_id|qty|    tag|
+------------+----------+------------+---+-------+
|Kemeja Batik|       101|           1|  2|pakaian|
|Kemeja Batik|       101|           1|  2| formal|
+------------+----------+------------+---+-------+

Total qty sebelum join : 12
Total qty setelah join : 24
== 5. Broadcast join: rencana eksekusi dengan dan tanpa hint ==
Tanpa hint (broadcast dimatikan):
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [pelanggan_id#4, pesanan_id#3, produk#5, qty#6, nama#1, kota#2]
   +- SortMergeJoin [pelanggan_id#4], [pelanggan_id#0], Inner
      :- Sort [pelanggan_id#4 ASC NULLS FIRST], false, 0
      :  +- Exchange hashpartitioning(pelanggan_id#4, 200), ENSURE_REQUIREMENTS, [plan_id=724]
      :     +- Filter isnotnull(pelanggan_id#4)
      :        +- Scan ExistingRDD[pesanan_id#3,pelanggan_id#4,produk#5,qty#6]
      +- Sort [pelanggan_id#0 ASC NULLS FIRST], false, 0
         +- Exchange hashpartitioning(pelanggan_id#0, 200), ENSURE_REQUIREMENTS, [plan_id=725]
            +- Filter isnotnull(pelanggan_id#0)
               +- Scan ExistingRDD[pelanggan_id#0,nama#1,kota#2]


Dengan broadcast(pelanggan):
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [pelanggan_id#4, pesanan_id#3, produk#5, qty#6, nama#1, kota#2]
   +- BroadcastHashJoin [pelanggan_id#4], [pelanggan_id#0], Inner, BuildRight, false, false
      :- Filter isnotnull(pelanggan_id#4)
      :  +- Scan ExistingRDD[pesanan_id#3,pelanggan_id#4,produk#5,qty#6]
      +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [plan_id=754]
         +- Filter isnotnull(pelanggan_id#0)
            +- Scan ExistingRDD[pelanggan_id#0,nama#1,kota#2]

Mari kita telaah baris-baris kode di atas:

Baris 38, 41, dan 44 (tiga join): pola join(tabel, kunci, tipe) memakai kunci pelanggan_id dengan tipe "inner", "left", dan "full_outer"; orderBy("pesanan_id") menjaga urutan baris agar perbandingan ketiga output mudah dibaca.
Baris 47-61 (fan-out): setiap produk memiliki tepat dua tag, sehingga join pesanan dengan tag pada kolom produk menempelkan dua baris tag ke setiap pesanan; filter pesanan_id == 101 memperlihatkan efek itu pada satu pesanan, lalu baris 58-61 menghitung sum("qty") sebelum dan sesudah join — angkanya berubah dari 12 menjadi 24 tanpa satu transaksi baru pun terjadi.
Baris 64 (spark.conf.set(..., -1)): konfigurasi spark.sql.autoBroadcastJoinThreshold dimatikan sementara agar rencana join pertama memakai strategi default yang paling umum.
Baris 66 dan 68 (dua explain()): join pertama tanpa hint menghasilkan SortMergeJoin dengan dua Exchange (shuffle); join kedua dibungkus broadcast(pelanggan) sehingga rencananya berubah menjadi BroadcastHashJoin. Angka plan_id pada output berubah-ubah antar eksekusi; struktur rencananya yang stabil dan itulah yang kita baca.

Inner, Left, dan Full Outer Join #

Ketiga output menampilkan tiga keputusan berbeda soal baris tanpa pasangan. Inner join membuang pesanan 106 karena pelanggan 99 tidak ada — hasilnya 7 baris. Left join mempertahankan seluruh pesanan dan mengisi kolom pelanggan dengan NULL — 8 baris. Full outer join juga memunculkan Agus yang tanpa pesanan — 9 baris. Satu detail kecil pada output ketiga: baris Agus tampil paling atas karena ORDER BY ascending Spark menempatkan NULL di urutan awal. Spark juga menyediakan right join (kebalikan left) serta left_semi dan left_anti yang hanya mempertahankan baris kiri yang punya atau tidak punya pasangan.

Tipe join	Baris tanpa pasangan	Jumlah baris hasil
`inner`	dibuang dari kedua sisi	7
`left`	sisi kiri dipertahankan, kanan jadi `NULL`	8
`full_outer`	kedua sisi dipertahankan, pasangan jadi `NULL`	9

Fan-out: Join yang Melipatgandakan Baris #

Bagian 4 adalah pelajaran paling mahal di chapter ini. Join dengan tabel tag melipatgandakan setiap pesanan menjadi dua baris, sehingga SUM(qty) ikut berlipat dua: 12 menjadi 24. Tidak ada galat yang muncul — angka 24 terlihat seperti omzet yang sah. Inilah fan-out, dan obatnya bukan menghindari join melainkan mengetahui grain masing-masing tabel: agregasi ke grain yang benar sebelum join, atau agregasi sesudah join dengan menyadari baris telah terduplikasi. Relasi banyak-ke-banyak seperti pesanan ke tag adalah penyebab paling umumnya di dunia nyata.

Broadcast Join: Menekan Shuffle #

Rencana pertama pada output bagian 5 memperlihatkan harga mahal join biasa: kedua tabel di-shuffle (dua Exchange) lalu diurutkan sebelum digabung. Rencana kedua memilih jalan lain: tabel pelanggan yang kecil dikirim utuh ke setiap executor (BroadcastExchange), sehingga pesanan tidak perlu berpindah dan tidak ada pengurutan. Secara default Spark sudah menyiarkan sisi yang ukurannya di bawah spark.sql.autoBroadcastJoinThreshold (10 MB); konfigurasi tersebut kami matikan di demo hanya untuk memperlihatkan perbedaannya. Hint broadcast() berguna ketika statistik ukuran tabel tidak akurat sehingga Spark salah menebak — dokumentasi lengkap threshold ada di halaman konfigurasi Spark . Catatan terakhir: baris AdaptiveSparkPlan isFinalPlan=false berarti AQE masih bisa mengubah rencana saat eksekusi berjalan — misalnya mengubah SortMergeJoin menjadi broadcast setelah ukuran shuffle asli diketahui.

Join menjawab pertanyaan data apa yang digabungkan; pertanyaan berikutnya adalah di mana data itu ditempatkan dan diproses. Kita masuk ke mekanisme di baliknya pada chapter Spark Partitioning.