Sembilan dari sepuluh menit durasi job agregasi TokoKita habis di satu task yang menunggu sendirian, sementara tiga task lainnya sudah lama selesai. Penyebabnya bukan bug — melainkan kombinasi dua mekanisme yang selalu ada di Spark: shuffle yang memindahkan data antar partition, dan skew yang membuat pemindahan itu tidak merata. Pada chapter ini kita membaca shuffle dari execution plan, membuktikan skew dengan kunci tamu yang menampung 80.000 dari 100.000 pesanan, lalu memperbaikinya dengan teknik salting dan agregasi dua tahap.

Apa itu Shuffle #

Transformasi Spark terbagi dua kelas. Narrow transformation seperti select dan filter memproses tiap partition secara mandiri — tidak ada data yang berpindah. Wide transformation seperti groupBy, join, dan orderBy menuntut semua baris berkunci sama berada di partition yang sama, sehingga Spark harus menyalin data antar executor lewat jaringan: peristiwa inilah shuffle, dan di execution plan tampak sebagai operator Exchange. Biayanya nyata: data di-serialize, ditulis ke disk lokal pengirim, ditransfer lewat jaringan, lalu dibaca kembali — semuanya memakan execution memory yang kita bahas di chapter Memory Management. Skew terjadi ketika satu kunci membawa jauh lebih banyak baris daripada kunci lain, sehingga satu partition hasil shuffle menjadi raksasa dan satu task menjadi penghambat seluruh job.

Demo: Kunci Timpang dan Salting #

Skrip berikut membangun 100 ribu pesanan yang 80%-nya berpelanggan tamu (checkout tanpa akun), lalu membandingkan sebaran partition sebelum dan sesudah salting:

python 
"""Shuffle, skew, dan salting pada kunci yang timpang."""
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

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

# 100.000 pesanan TokoKita: 80% checkout tanpa akun ("tamu"),
# sisanya tersebar merata ke lima pelanggan terdaftar.
# numPartitions=4 membuat sumber multi-partisi sehingga agregasi
# benar-benar mengalami shuffle.
pesanan = spark.range(100_000, numPartitions=4).select(
    F.col("id").alias("pesanan_id"),
    (F.when(F.col("id") < 80_000, F.lit("tamu"))
     .when((F.col("id") - 80_000) % 5 == 0, F.lit("Budi"))
     .when((F.col("id") - 80_000) % 5 == 1, F.lit("Siti"))
     .when((F.col("id") - 80_000) % 5 == 2, F.lit("Dewi"))
     .when((F.col("id") - 80_000) % 5 == 3, F.lit("Rina"))
     .otherwise(F.lit("Ayu"))).alias("pelanggan"),
    ((F.col("id") % 500) * 1000 + 15000).alias("nilai"),
)

print("=== Bagian 1: Exchange pada execution plan ===")
langsung = pesanan.groupBy("pelanggan").count()
langsung.explain()

print("=== Bagian 2: sebaran partisi pada kunci timpang ===")
sebaran_timpang = (pesanan.repartition(4, "pelanggan")
                   .withColumn("partisi", F.spark_partition_id())
                   .groupBy("partisi").count().orderBy("partisi"))
print("sebaran baris per partisi shuffle (sebelum salting):")
sebaran_timpang.show()

print("=== Bagian 3: salting dan agregasi dua tahap ===")
bergaram = pesanan.withColumn("garam", (F.col("pesanan_id") % 4))
sebaran_garam = (bergaram.repartition(4, "pelanggan", "garam")
                 .withColumn("partisi", F.spark_partition_id())
                 .groupBy("partisi").count().orderBy("partisi"))
print("sebaran baris per partisi shuffle (setelah salting):")
sebaran_garam.show()

tahap1 = bergaram.groupBy("garam", "pelanggan").count()
tahap2 = (tahap1.groupBy("pelanggan")
          .agg(F.sum("count").alias("jumlah"))
          .orderBy("pelanggan"))
print("hasil agregasi dua tahap (dengan salting):")
tahap2.show()
print("hasil agregasi langsung (tanpa salting):")
langsung.orderBy("pelanggan").show()

spark.stop()

Output yang Dihasilkan:

text 
=== Bagian 1: Exchange pada execution plan ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[pelanggan#2], functions=[count(1)])
   +- Exchange hashpartitioning(pelanggan#2, 4), ENSURE_REQUIREMENTS, [plan_id=15]
      +- HashAggregate(keys=[pelanggan#2], functions=[partial_count(1)])
         +- Project [CASE WHEN (id#0L < 80000) THEN tamu WHEN (((id#0L - 80000) % 5) = 0) THEN Budi WHEN (((id#0L - 80000) % 5) = 1) THEN Siti WHEN (((id#0L - 80000) % 5) = 2) THEN Dewi WHEN (((id#0L - 80000) % 5) = 3) THEN Rina ELSE Ayu END AS pelanggan#2]
            +- Range (0, 100000, step=1, splits=4)


=== Bagian 2: sebaran partisi pada kunci timpang ===
sebaran baris per partisi shuffle (sebelum salting):
+-------+-----+
|partisi|count|
+-------+-----+
|      0|88000|
|      1| 4000|
|      2| 8000|
+-------+-----+

=== Bagian 3: salting dan agregasi dua tahap ===
sebaran baris per partisi shuffle (setelah salting):
+-------+-----+
|partisi|count|
+-------+-----+
|      0|45000|
|      1| 3000|
|      2|23000|
|      3|29000|
+-------+-----+

hasil agregasi dua tahap (dengan salting):
+---------+------+
|pelanggan|jumlah|
+---------+------+
|      Ayu|  4000|
|     Budi|  4000|
|     Dewi|  4000|
|     Rina|  4000|
|     Siti|  4000|
|     tamu| 80000|
+---------+------+

hasil agregasi langsung (tanpa salting):
+---------+-----+
|pelanggan|count|
+---------+-----+
|      Ayu| 4000|
|     Budi| 4000|
|     Dewi| 4000|
|     Rina| 4000|
|     Siti| 4000|
|     tamu|80000|
+---------+-----+

Mari kita telaah baris-baris kode di atas:

  1. Baris 16-25 (konstruksi data timpang): id < 80_000 menghasilkan pelanggan tamu, sisanya bergilir ke lima nama terdaftar — pola khas data e-commerce nyata yang didominasi checkout tanpa akun.
  2. Baris 29 (langsung.explain()): membuktikan groupBy("pelanggan") memicu Exchange hashpartitioning(pelanggan, 4) — seluruh baris berkunci sama wajib berakhir di satu partition.
  3. Baris 32-34 (repartition(4, "pelanggan")): meniru partitioner shuffle lalu menandai tiap baris dengan nomor partitionnya (spark_partition_id), sehingga groupBy("partisi") mengukur sebaran sebenarnya.
  4. Baris 39 (withColumn("garam", ...)): salting — kunci tamu dipecah implisit menjadi tamu+garam 0-3 karena setiap baris membawa angka garam tambahan.
  5. Baris 40-42 (repartition(4, "pelanggan", "garam")): sekarang data disebar menurut pasangan kunci+garam; 80 ribu baris tamu terbagi ke empat kelompok.
  6. Baris 46-49 (agregasi dua tahap): tahap pertama mengagregasi per (garam, pelanggan) — grup kecil yang seimbang; tahap kedua menjumlahkan partial count per pelanggan untuk mengembalikan hasil sebenarnya.

Membaca Sebaran: 88000 versus 4000 #

Sebelum salting, partition 0 menampung 88.000 baris — kunci tamu (80.000) berbagi partition dengan sebagian kunci lain menurut hasil hash — sementara partition 1 hanya 4.000 baris dan satu partition bahkan tidak terisi. Rasio puncak 22 kali lipat berarti task termahal menentukan durasi job hampir sendirian. Setelah salting, sebaran menjadi [45000, 3000, 23000, 29000]: puncak turun hampir separuh karena 80 ribu baris tamu terbelah empat. Sebaran ini belum sempurna — penempatan pasangan kunci+garam tetap ditentukan hash, dan beberapa pasangan bisa berkerumup di partition yang sama — tetapi task terlama memendek signifikan. Pada cluster dengan ratusan partition, efeknya jauh lebih dramatis: kunci raksasa yang tadinya mengisi satu task kini terbagi ke banyak task.

Join Timpang: Broadcast dan AQE #

Skew juga menyerang join: satu kunci join raksasa membuat sort-merge join tertahan di satu task. Dua penanganan utamanya berbeda dari agregasi. Pertama, bila satu sisi join kecil (misalnya dimensi 8 cabang TokoKita), broadcast join menyalin seluruh sisi kecil ke semua executor sehingga tidak ada shuffle sama sekali — kita sudah melihat BroadcastHashJoin menggantikan SortMergeJoin pada chapter *Spark Joins. Kedua, sejak Spark 3, Adaptive Query Execution mendeteksi partition join yang berukuran jauh di atas rata-rata dan memecahnya otomatis (konfigurasi spark.sql.adaptive.skewJoin.enabled`, aktif secara default di Spark 4) — solusi built-in untuk skew join tanpa salting manual. Mekanisme AQE selengkapnya terdokumentasi di panduan performa SQL Spark .

Shuffle dan skew adalah biaya dari kekuatan Spark membagi data lintas mesin. Pertanyaan penutup section ini karenanya mendasar: kapan biaya tersebut memang layak dibayar — dan kapan kita sebaiknya memilih alat lain? Pertanyaan itulah yang dijawab tuntas pada chapter Kapan Menggunakan Spark.