KAL
Menu

Shuffle

Alasan shuffle mahal dan pemicunya.

Daftar isi

Shuffle adalah proses mendistribusikan ulang data lintas partisi sehingga baris dengan kunci yang sama berkumpul di partisi yang sama, dan proses ini adalah biaya terbesar yang paling sering muncul pada pekerjaan Spark. Topik ini membahas mengapa shuffle mahal, operasi apa saja yang memicunya, dan cara membuktikan perpindahan partisi itu pada data kecil. Setelah membacanya, pemicu shuffle dapat dikenali dari kode, dan alasan lambatnya sebuah stage dapat ditelusuri ke pertukaran data ini.

Mengapa shuffle mahal#

Operasi tanpa shuffle dapat dihitung partisi demi partisi secara mandiri, sedangkan operasi dengan shuffle menuntut setiap partisi keluaran menerima kontribusi dari semua partisi masukan. Konsekuensinya berlapis. Data harus ditulis ke disk di sisi pengirim, dikirim melalui jaringan antar pekerja, lalu dibaca kembali di sisi penerima, sehingga shuffle memakai tiga sumber daya sekaligus, yaitu disk, jaringan, dan memori untuk penyanggaan. Pada cluster nyata, inilah bagian yang paling sering menjadi kemacetan pada sebuah stage.

Biayanya juga tidak proporsional dengan ukuran data semata. Jumlah berkas sementara, task, dan koneksi jaringan tumbuh mengikuti jumlah partisi, sehingga shuffle dengan partisi berlebihan memakan waktu pada penjadwalan, bukan pada pemindahan byte. Penyesuaian spark.sql.shuffle.partitions yang dikenalkan pada topik agregasi adalah pengendali langsung atas dimensi ini, dan topik esensi penyetelan mengulangnya dari sisi produksi.

Ada juga sisi positif yang perlu dipahami, yaitu shuffle menulis data ke disk secara terkelola, sehingga pekerjaan dengan data yang jauh melampaui memori tetap dapat selesai. Mahal bukan berarti salah, melainkan harus dibenarkan oleh kebutuhan operasinya.

Operasi yang memicu shuffle#

Daftar pemicunya menyangkut semua operasi yang maknanya bergantung pada pengelompokan atau pengurutan global. groupBy dan agregasinya membutuhkan kunci kelompok yang terkumpul, orderBy membutuhkan urutan global, join menuntut baris dengan kunci yang sama bertemu, dan repartition secara eksplisit mendistribusikan ulang data. Perubahan bentuk seperti distinct dan operasi window dengan partitionBy juga termasuk daftar ini.

Sebaliknya, filter, select, withColumn, dan sample tidak memicu shuffle karena setiap partisi dapat diproses sendiri. Kebiasaan membaca kode dengan lensa ini, yaitu memisahkan operasi lokal dari operasi lintas partisi, memungkinkan biaya sebuah pipeline diprediksi sebelum pekerjaan dijalankan. Batas stage pada rencana explain() adalah konfirmasi formalnya, karena simpul Exchange selalu berdiri di tempat shuffle terjadi.

Membuktikan perpindahan partisi#

Fungsi F.spark_partition_id() menandai setiap baris dengan nomor partisinya, sehingga perpindahan baris antar partisi dapat diamati langsung pada data kecil. Skrip berikut membandingkan letak baris sebelum dan sesudah groupBy, termasuk jumlah partisi hasil yang dilaporkan Spark.

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

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("peraga-shuffle")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

df = spark.createDataFrame(
    [(1, "Ali", "10A", 85), (2, "Budi", "10A", 90), (3, "Citra", "10B", 78), (4, "Dewi", "10B", 92)],
    "id long, nama string, kelas string, nilai long",
)

print("=== id partisi sebelum shuffle ===")
df.withColumn("partisi", F.spark_partition_id()).orderBy("partisi", "id").show()

print("=== id partisi setelah groupBy (shuffle) ===")
hasil = df.groupBy("kelas").count()
hasil.withColumn("partisi", F.spark_partition_id()).orderBy("partisi", "kelas").show()
print("jumlah partisi hasil:", hasil.rdd.getNumPartitions())

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== id partisi sebelum shuffle ===
+---+-----+-----+-----+-------+
| id| nama|kelas|nilai|partisi|
+---+-----+-----+-----+-------+
|  1|  Ali|  10A|   85|      0|
|  2| Budi|  10A|   90|      0|
|  3|Citra|  10B|   78|      1|
|  4| Dewi|  10B|   92|      1|
+---+-----+-----+-----+-------+

=== id partisi setelah groupBy (shuffle) ===
+-----+-----+-------+
|kelas|count|partisi|
+-----+-----+-------+
|  10A|    2|      0|
|  10B|    2|      0|
+-----+-----+-------+

jumlah partisi hasil: 1

Keluaran pertama memperlihatkan pembagian awal yang mengikuti pemotongan data saat dibuat, yaitu dua baris pertama di partisi 0 dan dua baris sisanya di partisi 1. Keluaran kedua menunjukkan kedua kelompok kini bersama di partisi yang sama, dan angka jumlah partisi hasil: 1 menegaskan bahwa AQE memadatkan partisi hasil shuffle yang kecil menjadi satu, karena dua partisi yang masing-masing hanya memuat satu baris hasil terlalu boros untuk dijadwalkan. Perpadatan ini adalah perilaku AQE yang normal, dan jumlah partisi yang diminta tetap dua sesuai konfigurasi, tetapi rencana final menyesuaikannya dengan ukuran data nyata.

Kapan shuffle diterima#

Pertanyaan yang tepat bukanlah bagaimana menghindari shuffle sepenuhnya, karena agregasi dan join memang memerlukannya, melainkan bagaimana tidak melakukan shuffle yang tidak diperlukan. Pola yang mahal antara lain orderBy di tengah pipeline padahal urutan hanya dibutuhkan pada keluaran akhir, repartition yang berulang tanpa kebutuhan, dan join antar tabel besar yang sisi kecilnya sebenarnya bisa di-broadcast. Penanganan tiap pola ini dibahas pada topik partisi dan esensi penyetelan.

Ukuran keberhasilannya sederhana, yaitu setiap Exchange pada rencana harus dapat dijelaskan oleh kebutuhan semantiknya. Rencana dengan Exchange yang sedikit dan data yang seimbang adalah tanda pipeline yang sehat, dan perbandingan rencana sebelum-sesudah perubahan kode adalah cara membuktikannya.

Kesalahan umum#

Kesalahan pertama adalah menyalahkan shuffle untuk semua kelambatan. Penyebab lain seperti partisi timpang atau kelambatan menampilkan data sering tercampur, dan pembacaan Spark UI per stage adalah cara memisahkannya. Shuffle yang sehat terlihat sebagai stage yang durasinya merata di semua task.

Kesalahan kedua adalah menetapkan spark.sql.shuffle.partitions yang sangat besar dengan asumsi lebih banyak partisi selalu lebih cepat. Partisi yang isinya terlalu kecil membebani penjadwalan dan menumpuk berkas shuffle kecil, sedangkan partisi yang terlalu sedikit membuat task besar yang rentan kehabisan memori. Nilainya dipilih dari ukuran data, dan AQE membantu memperbaikinya saat berjalan.

Kesalahan ketiga adalah melakukan orderBy di tengah rantai demi kenyamanan membaca, lalu mengurutkan lagi di akhir. Pengurutan pertama adalah shuffle penuh yang hasilnya terbuang. Karena itu, urutkan hanya pada action terakhir atau pada bagian yang benar-benar memerlukannya.

Ringkasan#

  • Shuffle mempertemukan baris dengan kunci yang sama lintas partisi dan membayar biayanya pada disk, jaringan, dan memori sekaligus.
  • Pemicunya adalah groupBy, orderBy, join, distinct, repartition, dan window dengan partitionBy, sedangkan filter dan select tetap lokal.
  • F.spark_partition_id() membuktikan perpindahan partisi pada data kecil, dan AQE memadatkan partisi hasil shuffle yang terlalu kecil.
  • Tujuannya bukan menghapus shuffle, melainkan memastikan setiap Exchange pada rencana dibenarkan oleh kebutuhan operasinya.

Referensi#