KAL
Menu

Model memori

Tata letak memori executor, spill, intuisi OOM.

Daftar isi

Error out of memory adalah gangguan paling umum ketika aplikasi Spark berpindah dari data kecil ke data nyata, dan sebagian besar kebingungannya hilang setelah tata letak memori dipahami. Topik ini membahas bagaimana Spark membagi heap yang tersedia, mengapa pembagian itu menentukan batas agregasi dan pengurutan, dan bagaimana spill menyelamatkan pekerjaan yang datanya lebih besar dari memori.

Pembagian heap pada model unified#

Sejak Spark 1.6, memori eksekusi dan memori penyimpanan berbagi satu kawasan bernama unified memory. Dari heap yang tersedia, Spark menyisihkan sekitar tiga ratus MB sebagai cadangan sistem, lalu kawasan unified tersebut dihitung dari sisanya sebagai persentase yang ditentukan spark.memory.fraction dengan nilai bawaan 0.6. Kawasan ini dipakai bersama oleh dua pemakai, yaitu eksekusi untuk shuffle, join, sort, dan agregasi, serta penyimpanan untuk cache dan variabel broadcast.

Bagian dalam kawasan itu dibagi lagi menurut spark.memory.storageFraction dengan nilai bawaan 0.5, yang menandakan jatah penyimpanan. Sifat berbagi inilah yang penting, karena kedua pemakai boleh saling meminjam. Ketika cache kosong, eksekusi boleh memakai hampir seluruh kawasan, dan ketika data besar di-cache, penyimpanan dapat merebut kembali jatahnya dengan menyingkirkan blok yang tidak sedang dipakai. Model ini menghilangkan dinding kaku antara kedua pemakai pada model lama, dan menutup keluhan umum berupa cache yang menganggur sementara eksekusi kehabisan memori.

Skrip berikut membaca konfigurasi yang berlaku pada sesi lokal dan menghitung estimasi tata letaknya, sehingga angka-angka di atas menjadi nyata pada heap yang benar-benar dipakai.

python peraga_memori.py 
from pyspark.sql import SparkSession

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

sc = spark.sparkContext
conf = sc._conf

heap_mb = sc._jvm.java.lang.Runtime.getRuntime().maxMemory() // (1024 * 1024)
cadangan = 300
unified_mb = (heap_mb - cadangan) * 60 // 100
penyimpanan_mb = unified_mb * 50 // 100

print("=== konfigurasi memori yang terbaca Spark ===")
print("master             :", conf.get("spark.master"))
print("driver memory      :", conf.get("spark.driver.memory", "1g"))
print("memory fraction    :", conf.get("spark.memory.fraction", "0.6"))
print("storage fraction   :", conf.get("spark.memory.storageFraction", "0.5"))
print("=== estimasi tata letak heap ===")
print("heap tersedia      :", heap_mb, "MB")
print("unified memory     :", unified_mb, "MB")
print("bagian storage     :", penyimpanan_mb, "MB")
print("bagian execution   :", unified_mb - penyimpanan_mb, "MB")

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== konfigurasi memori yang terbaca Spark ===
master             : local[2]
driver memory      : 1g
memory fraction    : 0.6
storage fraction   : 0.5
=== estimasi tata letak heap ===
heap tersedia      : 1024 MB
unified memory     : 434 MB
bagian storage     : 217 MB
bagian execution   : 217 MB

Angka pada keluaran membuka cara berpikir yang benar. Dari satu giga heap, sekitar tiga ratus MB menjadi cadangan, lalu enam puluh persen sisanya menjadi kawasan unified sekitar empat ratus tiga puluh MB, dan setengahnya adalah jatah penyimpanan. Ketika sebuah agregasi diberi data yang jauh lebih besar dari ratusan MB kawasan eksekusi, ia tidak langsung gagal, melainkan beralih ke mekanisme berikutnya, yaitu spill.

Spill dan batas memori#

Spill adalah penulisan sementara data ke disk ketika kawasan eksekusi tidak cukup, misalnya saat struktur pengurutan atau tabel agregasi melebihi jatah. Hasil pekerjaan tetap benar, hanya lebih lambat, karena data yang sama ditulis dan dibaca kembali dari disk. Pada Spark UI, jumlah spill memori dan disk terlihat per stage, dan nilai yang besar pada pekerjaan yang seharusnya ringan adalah sinyal utama bahwa partisi terlalu besar.

Intuisi OOM karenanya perlu dikoreksi. Kegagalan memori jarang disebabkan oleh data yang besar dalam arti seluruh berkas, melainkan oleh partisi tunggal yang terlalu besar, karena eksekusi bekerja per partisi. Dua penyebab paling lazim adalah pengurutan atau agregasi pada kunci yang timpang sehingga satu partisi menampung jauh lebih banyak baris dari lainnya, serta collect() atau broadcast yang menarik terlalu banyak data ke driver. Perbaikannya pun konsisten, yaitu perbaiki kunci agar distribusinya lebih rata, kurangi ukuran sisi yang di-broadcast, dan hindari menarik seluruh hasil ke driver.

Satu koreksi tambahan penting untuk mode lokal. spark.driver.memory yang diset melalui spark-submit tidak berlaku ketika sesi dibangun dari PySpark biasa, karena JVM sudah berjalan saat konfigurasi itu terbaca. Pada mode lokal, ukuran heap ditetapkan saat proses JVM diluncurkan, dan inilah alasan angka heap pada keluaran di atas tetap satu giga meskipun konfigurasi spark.driver.memory diubah setelah JVM berjalan.

Intuisi menyetel memori#

Menyetel memori berangkat dari dua pertanyaan, yaitu seberapa besar partisi terbesar dan seberapa banyak data yang perlu tinggal di memori sebagai cache. Jawaban pertama mengarahkan ukuran partisi, dan memperkecilnya cukup dengan menaikkan jumlah partisi shuffle, bukan dengan menaikkan memori. Jawaban pertanyaan kedua menentukan apakah penambahan memori benar-benar dibutuhkan, karena cache yang jauh melebihi jatah penyimpanan akan mengeluarkan blok lama dan menurunkan manfaat cache itu.

Pada cluster, angka pada skrip di atas berlaku per executor, bukan per aplikasi. Executor dengan --executor-memory empat giga memiliki kawasan unified sekitar dua giga dua ratus MB menurut rumus yang sama, dan seluruh task di executor itu berbagi kawasan yang sama. Memahami satuan ini mencegah kesalahan umum berupa mengira total memori aplikasi sebagai penjumlahan memori seluruh executor, padahal batas yang menentukan kegagalan task adalah memori pada satu executor saja.

Kesalahan umum#

Kesalahan pertama adalah mengobati OOM dengan menaikkan memori tanpa melihat distribusi partisi. Partisi timpang akan tetap timpang, dan memori yang naik hanya menunda kegagalan. Perbaikan yang tahan lama adalah memeriksa kunci agregasi atau join, lalu menyusun partisi agar lebih rata, termasuk menyaring kolom dan baris yang tidak diperlukan sejak pembacaan.

Kesalahan kedua adalah mem-cache data mentah yang besar padahal hanya beberapa kolom yang dipakai berulang. Cache menelan jatah penyimpanan, menyingkirkan blok lain, dan sering kali membuat pekerjaan lebih lambat daripada membaca ulang berkas yang sudah efisien. Aturan praktisnya, cache hasil antara yang mahal dan kecil, bukan seluruh data mentah.

Kesalahan ketiga adalah menaikkan spark.driver.memory di dalam kode atau di spark-defaults.conf lalu bingung karena tidak berpengaruh. Pada PySpark, ukuran heap driver ditetapkan saat proses JVM dimulai, sehingga cara yang bekerja adalah meluncurkan melalui spark-submit dengan opsi --driver-memory, atau mengatur variabel PYSPARK_SUBMIT_ARGS sebelum sesi dibangun. Topik spark-submit pada modul ini memeragakan jalur peluncuran tersebut.

Ringkasan#

  • Heap dibagi menjadi cadangan sistem dan kawasan unified sebesar spark.memory.fraction dari sisanya, dan kawasan itu dipakai bersama oleh eksekusi serta penyimpanan.
  • Jatah penyimpanan hanya nilai minimum, karena kedua pemakai saling meminjam dan penyimpanan dapat merebut kembali jatahnya dari blok yang tidak terpakai.
  • Spill menjaga kebenaran hasil dengan menulis sementara ke disk, dan spill besar menandai partisi yang terlalu besar alih-alih kekurangan memori total.
  • Batas yang menentukan kegagalan adalah memori pada satu executor dan satu partisi, sehingga distribusi partisi lebih penting daripada angka memori total.

Referensi#