KAL
Menu

Esensi penyetelan

Partisi shuffle, broadcast join, AQE.

Daftar isi

Penyetelan Spark yang paling banyak berdampak justru yang paling sederhana, yaitu menjaga jumlah dan ukuran partisi shuffle wajar, memaksa join dengan tabel kecil menjadi broadcast, dan membiarkan adaptive query execution memperbaiki rencana saat berjalan. Topik ini merangkum ketiganya dengan bukti rencana eksekusi nyata, sehingga keputusan penyetelan berhenti menjadi tebakan.

Partisi shuffle sebagai satuan kerja#

Setiap shuffle membagi data ke sejumlah partisi yang ditentukan spark.sql.shuffle.partitions, dengan nilai bawaan dua ratus. Jumlah ini menentukan banyaknya satuan kerja paralel pada stage berikutnya, sehingga terlalu sedikit membuat sebagian core menganggur dan setiap partisi membengkak, sedangkan terlalu banyak memecah kerja menjadi ribuan task kecil yang biaya penjadwalannya melebihi manfaatnya. Pada data kecil dan menengah, angka dua ratus jelas terlalu besar, dan itulah sebabnya seluruh skrip pada jalur ini menyetelnya ke angka kecil.

Ukuran partisi yang sehat pada satu executor berada di kisaran seratus MB sampai beberapa ratus MB. Perkiraannya dapat dihitung dengan rumus sederhana, yaitu volume data setelah shuffle dibagi jumlah partisi. Ketika partisi terbesar jauh melebihi kisaran itu, risikonya spill dan kegagalan memori sebagaimana dijelaskan pada topik model memori, sedangkan partisi yang jauh di bawahnya memboroskan biaya per task. Penghitungan ini juga yang membuat AQE begitu berharga, karena AQE menggabungkan partisi kecil secara otomatis tanpa perlu tebakan manual.

Pemilihan jumlah partisi yang baik mengikuti dua hal, yaitu jumlah core yang benar-benar tersedia dan bentuk kuncinya. Pada mode lokal dua core, empat sampai delapan partisi sudah melampaui kebutuhan, dan modul cara Spark berjalan memeragakan perilakunya. Pada cluster, jumlah partisi biasanya kelipatan jumlah core total, dengan sedikit kelebihan untuk menyeimbangkan partisi yang tidak rata.

Broadcast join dengan bukti rencana#

Join antara tabel besar dan tabel kecil adalah kasus yang paling mahal bila dibiarkan shuffle, dan paling menguntungkan bila menjadi broadcast. Skrip berikut menampilkan kedua rencana pada data yang sama, yaitu join biasa dan join dengan hint F.broadcast(), sehingga perbedaannya terlihat di rencana fisik.

python penyetelan_join.py 
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("penyetelan-join")
    .config("spark.sql.shuffle.partitions", "4")
    .config("spark.sql.adaptive.enabled", "false")
    .getOrCreate()
)

fakta = spark.createDataFrame(
    [(i, f"kamar{i % 3}", 20.0 + (i % 5)) for i in range(8)],
    "id long, kamar string, suhu double",
)
label = spark.createDataFrame(
    [("kamar0", "lantai-1"), ("kamar1", "lantai-2"), ("kamar2", "lantai-3")],
    "kamar string, lantai string",
)

print("=== rencana join default ===")
fakta.join(label, on="kamar").explain()

print("=== rencana dengan broadcast hint ===")
fakta.join(F.broadcast(label), on="kamar").explain()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== rencana join default ===
== Physical Plan ==
*(5) Project [kamar#1, id#0L, suhu#2, lantai#4]
+- *(5) SortMergeJoin [kamar#1], [kamar#3], Inner
   :- *(2) Sort [kamar#1 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(kamar#1, 4), ENSURE_REQUIREMENTS, [plan_id=31]
   :     +- *(1) Filter isnotnull(kamar#1)
   :        +- *(1) Scan ExistingRDD[id#0L,kamar#1,suhu#2]
   +- *(4) Sort [kamar#3 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(kamar#3, 4), ENSURE_REQUIREMENTS, [plan_id=37]
         +- *(3) Filter isnotnull(kamar#3)
            +- *(3) Scan ExistingRDD[kamar#3,lantai#4]


=== rencana dengan broadcast hint ===
== Physical Plan ==
*(2) Project [kamar#1, id#0L, suhu#2, lantai#4]
+- *(2) BroadcastHashJoin [kamar#1], [kamar#3], Inner, BuildRight, false, false
   :- *(2) Filter isnotnull(kamar#1)
   :  +- *(2) Scan ExistingRDD[id#0L,kamar#1,suhu#2]
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=86]
      +- *(1) Filter isnotnull(kamar#3)
         +- *(1) Scan ExistingRDD[kamar#3,lantai#4]

Rencana default memperlihatkan SortMergeJoin dengan dua Exchange hashpartitioning, yaitu kedua sisi harus di-shuffle, diurutkan, lalu digabungkan. Rencana dengan hint menggantinya menjadi BroadcastHashJoin, dan yang dikirim hanyalah sisi label melalui satu BroadcastExchange, sementara sisi fakta tidak berpindah partisi sama sekali dan tidak perlu sortir. Pada data nyata, menghilangkan satu shuffle besar per kueri adalah penyetelan yang paling menghemat waktu.

Menariknya, Spark tidak otomatis memilih broadcast pada rencana awal di atas meskipun kedua tabelnya kecil, karena estimasi ukuran data dalam memori belum tentu akurat. Di sinilah AQE berperan saat eksekusi, dan ketika AQE dinonaktifkan seperti pada skrip, rencana awal itulah yang dieksekusi. Untuk sisi yang ukurannya diketahui kecil dengan pasti, menyatakan hint secara eksplisit tetap paling dapat diandalkan, dan ambang otomatisnya dapat disetel melalui spark.sql.autoBroadcastJoinThreshold.

Mengamati partisi hasil join#

Jumlah partisi hasil shuffle dapat diamati langsung dengan spark_partition_id(). Skrip berikut menghitung baris per partisi pada hasil join default di atas, dengan AQE dimatikan agar pemadatan partisi otomatis tidak menutupi pola aslinya.

python distribusi_join.py 
# tidak-dijalankan
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("distribusi-join")
    .config("spark.sql.shuffle.partitions", "2")
    .config("spark.sql.adaptive.enabled", "false")
    .getOrCreate()
)

fakta = spark.createDataFrame(
    [(i, f"kamar{i % 3}", 20.0 + (i % 5)) for i in range(8)],
    "id long, kamar string, suhu double",
)
label = spark.createDataFrame(
    [("kamar0", "lantai-1"), ("kamar1", "lantai-2"), ("kamar2", "lantai-3")],
    "kamar string, lantai string",
)

hasil = fakta.join(label, on="kamar")
hasil.groupBy(F.spark_partition_id().alias("partisi")).count().orderBy("partisi").show()

spark.stop()
text Ilustrasi keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
+-------+-----+
|partisi|count|
+-------+-----+
|      0|    3|
|      1|    5|
+-------+-----+

Dari dua partisi yang diminta, keduanya terisi dengan pembagian tiga dan lima baris, dan tidak ada partisi kosong karena kuncinya hanya tiga nilai. Pada kunci yang lebih banyak, partisi kosong atau timpang kerap muncul, dan modul streaming memeragakan perilaku serupa pada agregasi ber-window. Skrip ini berdurasi dekat batas verifikasi otomatis pada sandbox dua core sehingga keluarannya diambil dari pengujian manual yang skripnya identik, mengikuti pola yang sama dengan topik UDF dan watermark.

Dengan AQE menyala, pemadatan partisi kecil akan mengubah gambaran ini saat eksekusi, dan jumlah partisi akhir bisa lebih kecil daripada angka yang diminta. Karena itu, pengamatan seperti di atas paling informatif ketika dilakukan dua kali, sekali dengan AQE mati untuk melihat rencana mentah, dan sekali dengan AQE menyala untuk melihat hasil penyesuaiannya.

Kesalahan umum#

Kesalahan pertama adalah menyetel spark.sql.shuffle.partitions sekali untuk seluruh pekerjaan tanpa melihat volumenya. Angka yang pas untuk agregasi ringan bisa terlalu kecil untuk join data besar, dan konfigurasi ini layak disetel per tahap pekerjaan melalui spark.conf.set() ketika satu pipeline memuat tahap-tahap dengan volume berbeda.

Kesalahan kedua adalah mem-broadcast tabel yang sebenarnya besar berdasarkan tebakan, bukan berdasarkan ukuran. Seluruh sisi dikirim ke setiap executor dan disalin di memori masing-masing, sehingga pekerjaan gagal atau melambat. Keputusan broadcast sebaiknya berdasarkan ukuran berkas yang dimampatkan ditambah perkiraan inflasi format, dan bila ragu, biarkan estimasi Spark atau AQE yang memutuskan.

Kesalahan ketiga adalah menonaktifkan AQE karena rencananya tampak lebih panjang, lalu menulis ulang logika untuk menggantikan kerjanya. AQE hanya mengubah keputusan yang datanya sudah dikenal saat eksekusi, dan mematikannya berarti menerima estimasi buta sejak awal. Kasus yang sah untuk mematikannya justru pengamatan seperti skrip di atas, di mana rencana mentah perlu dilihat tanpa penyesuaian otomatis.

Ringkasan#

  • spark.sql.shuffle.partitions menentukan banyaknya satuan kerja paralel setelah shuffle, dan nilai yang sehat mengikuti volume data serta jumlah core, bukan nilai bawaan dua ratus.
  • Broadcast join menghapus shuffle dan sortir pada sisi besar, dan hint F.broadcast() memaksanya meski estimasi otomatis tidak memilihnya.
  • Rencana fisik adalah bukti keputusan penyetelan, dan perbedaan SortMergeJoin dengan dua Exchange versus BroadcastHashJoin dengan satu BroadcastExchange terbaca langsung dari explain().
  • AQE memperbaiki rencana saat eksekusi dengan menggabungkan partisi kecil dan memilih strategi berdasarkan statistik nyata, dan mematikannya hanya untuk pengamatan rencana mentah.

Referensi#