Memory Management

Job orderBy Budi tiba-tiba lambat ketika datanya naik ke 4 juta baris, padahal kode tidak berubah. Di balik kelambatan itu ada drama memori: executor kehabisan ruang kerja, lalu Spark terpaksa menumpahkan data ke disk — peristiwa yang disebut spill. Untuk membaca gejala seperti ini kita perlu memahami cara Spark membagi memori executor: berapa jatah untuk komputasi, berapa untuk cache, dan kapan batas itu terlampaui. Pada chapter ini kita hitung pembagian memori dari angka nyata sesi Spark, lalu sengaja memaksa spill dengan heap kecil dan mengukur akibatnya.

Unified Memory Manager #

Sejak Spark 1.6, memori executor dikelola unified memory manager. Dari total heap JVM, sekitar 300 MB selalu disisihkan untuk kebutuhan sistem Spark sendiri. Sisa heap dikalikan spark.memory.fraction (default 0.6) menjadi unified pool, yang dibagi lagi menjadi dua wilayah: execution memory untuk memori kerja shuffle, join, sort, dan agregasi; serta storage memory untuk block cache hasil cache(). Batas antar wilayah ini lunak — execution boleh meminjam storage yang kosong dan sebaliknya, dengan aturan evict: storage boleh menyingkirkan block cache lama demi execution, tetapi execution tidak pernah direbut paksa oleh storage. Rumus dan angka default ini terdokumentasi di tuning guide Spark .

Skrip Pengukuran dan Hasilnya #

Ukuran heap JVM hanya bisa diatur sebelum JVM hidup, jadi pengukuran spill berjalan sebagai proses Python terpisah. Skrip utama menghitung pool memori sesi berjalan, membandingkan efek spark.sql.shuffle.partitions dengan AQE nyala-mati, lalu memanggil pekerja terpisah:

python
"""Konfigurasi memori Spark: pool unified, shuffle partitions, dan spill."""
import subprocess
import sys
from pathlib import Path

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

RESERVED_MB = 300  # memori yang selalu disisihkan (lihat tuning guide Spark)

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

print("=== Bagian 1: angka memori sesi ini (default) ===")
# 0.6 dan 300 MB adalah nilai default terdokumentasi di tuning guide Spark;
# sesi ini tidak mengubahnya, jadi pool dihitung dari angka default tersebut.
FRAKSI_DEFAULT = 0.6
heap = spark._jvm.java.lang.Runtime.getRuntime().maxMemory()
unified = (heap - RESERVED_MB * 1024**2) * FRAKSI_DEFAULT
print(f"heap maksimum JVM           : {heap / 1024**2:.0f} MB")
print(f"pool unified (exec+storage) : {unified / 1024**2:.0f} MB = "
      f"(heap - {RESERVED_MB} MB) x {FRAKSI_DEFAULT}")

print("=== Bagian 2: efek spark.sql.shuffle.partitions ===")
# numPartitions=4 memastikan sumber memiliki beberapa partisi sehingga
# agregasi benar-benar melakukan shuffle (bukan dioptimasi jadi satu partisi).
pesanan = spark.range(100_000, numPartitions=4).select(
    (F.col("id") % 8).alias("cabang"),
    ((F.col("id") % 500) * 1000 + 15000).cast("double").alias("nilai"),
)
for aqe in ("true", "false"):
    spark.conf.set("spark.sql.adaptive.enabled", aqe)
    for jumlah in (8, 2):
        spark.conf.set("spark.sql.shuffle.partitions", str(jumlah))
        hasil = pesanan.groupBy("cabang").agg(F.sum("nilai").alias("total"))
        n = hasil.rdd.getNumPartitions()
        print(f"AQE={aqe:<5} shuffle.partitions={jumlah} -> {n} partisi")
spark.conf.set("spark.sql.adaptive.enabled", "true")

print("=== Bagian 3: spill pada proses terpisah (heap 512 MB) ===")
pekerja = Path(__file__).parent / "ch10_spill_worker.py"
jalankan = subprocess.run([sys.executable, str(pekerja)],
                          capture_output=True, text=True, timeout=240)
print(jalankan.stdout, end="")
if jalankan.returncode != 0:
    print(jalankan.stderr, file=sys.stderr)
    sys.exit(jalankan.returncode)

spark.stop()

Pekerja terpisah inilah bintang pengukurannya — heap dikecilkan ke 512 MB dan pool unified diperas hingga tersisa 10%:

python
"""Pekerja ch10: memaksa spill dengan heap 512 MB dan pool eksekusi kecil.

Dijalankan sebagai proses terpisah oleh ch10_memory.py karena ukuran heap
JVM hanya bisa diatur sebelum JVM dimulai.
"""
import json
import os
import time
import urllib.request

os.environ.setdefault(
    "PYSPARK_SUBMIT_ARGS", "--driver-memory 512m pyspark-shell")

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .master("local[1]")
         .config("spark.ui.enabled", "true")      # REST API untuk metrik
         .config("spark.ui.port", "0")
         .config("spark.sql.shuffle.partitions", "2")
         .config("spark.memory.fraction", "0.1")  # pool eksekusi dibuat kecil
         .getOrCreate())

heap = spark._jvm.java.lang.Runtime.getRuntime().maxMemory()
print(f"heap maksimum JVM    : {heap / 1024**2:.0f} MB")
print(f"spark.memory.fraction: {spark.conf.get('spark.memory.fraction')}")

pesanan = spark.range(4_000_000).select(
    F.col("id").alias("pesanan_id"),
    ((F.col("id") % 100_000) * 7 + 15000).alias("nilai"),
)

t0 = time.perf_counter()
pesanan.orderBy("nilai").write.format("noop").mode("overwrite").save()
durasi = time.perf_counter() - t0
print(f"orderBy 4 juta baris : {durasi:.1f} detik")

ui = spark.sparkContext.uiWebUrl
app_id = spark.sparkContext.applicationId
stages = json.load(urllib.request.urlopen(
    f"{ui}/api/v1/applications/{app_id}/stages"))
mem_spill = sum(s.get("memoryBytesSpilled", 0) for s in stages)
disk_spill = sum(s.get("diskBytesSpilled", 0) for s in stages)
print(f"memoryBytesSpilled   : {mem_spill / 1024**2:.1f} MB")
print(f"diskBytesSpilled     : {disk_spill / 1024**2:.1f} MB")

spark.stop()

Output yang Dihasilkan:

text
=== Bagian 1: angka memori sesi ini (default) ===
heap maksimum JVM           : 1024 MB
pool unified (exec+storage) : 434 MB = (heap - 300 MB) x 0.6
=== Bagian 2: efek spark.sql.shuffle.partitions ===
AQE=true  shuffle.partitions=8 -> 1 partisi
AQE=true  shuffle.partitions=2 -> 1 partisi
AQE=false shuffle.partitions=8 -> 8 partisi
AQE=false shuffle.partitions=2 -> 2 partisi
=== Bagian 3: spill pada proses terpisah (heap 512 MB) ===
heap maksimum JVM    : 512 MB
spark.memory.fraction: 0.1
orderBy 4 juta baris : 3.8 detik
memoryBytesSpilled   : 120.0 MB
diskBytesSpilled     : 43.2 MB

Mari kita telaah baris-baris kode di atas:

Baris 22-26 (perhitungan pool): Runtime.maxMemory() membaca heap maksimum JVM dari dalam sesi; pool unified dihitung manual dari rumus resmi (heap - 300 MB) × 0.6 karena Spark tidak mengekspos angkanya langsung.
Baris 35-42 (matriks AQE × shuffle.partitions): mengukur jumlah partition hasil agregasi pada empat kombinasi konfigurasi — bukti bahwa AQE menggabungkan partition kecil tanpa peduli konfigurasi awalnya.
Baris 46-47 (subprocess.run): menjalankan pekerja sebagai proses baru; capture_output=True memisahkan stdout/stderr, dan kode keluar tidak nol diteruskan sebagai kegagalan.
Baris 11-12 pekerja (PYSPARK_SUBMIT_ARGS): satu-satunya cara mengatur --driver-memory dari dalam PySpark — variabel ini dibaca saat JVM baru dibuat, itulah alasannya pekerja harus proses terpisah.
Baris 22 pekerja (spark.memory.fraction=0.1): memangkas pool unified menjadi 10% dari biasanya, mensimulasikan executor dengan memori kerja sempit.
Baris 35 pekerja (write.format("noop")): menjalankan orderBy penuh sampai selesai tanpa benar-benar menulis hasil ke disk — sink “tanpa operasi”.
Baris 39-46 pekerja (REST API Spark UI): membaca metrik memoryBytesSpilled dan diskBytesSpilled per stage dari endpoint /api/v1/applications/.../stages.

Membaca Angka Spill #

Hasil bagian 1 sesuai rumus: heap 1024 MB dikurangi 300 MB sistem, dikali 0.6, menghasilkan pool unified 434 MB. Pada pekerja, heap 512 MB dengan spark.memory.fraction 0.1 menyisakan pool unified sekitar 21 MB — sengaja dibuat jauh lebih kecil daripada kebutuhan orderBy 4 juta baris. Akibatnya terbaca jelas: 120,0 MB data kerja ter-spill dari memori, dan 43,2 MB di antaranya benar-benar ditulis ke disk (lebih kecil karena terkompresi). Durasi 3,8 detik pada run ini (run lain di mesin yang sama mencatat 4,1 detik — angka waktu memang bergerak), sedangkan kedua angka spill identik antar run karena ditentukan struktur datanya, bukan kecepatan mesin. Spill bukan kegagalan — Spark menyelesaikan pekerjaannya dengan benar — tetapi durasinya bisa berkali lipat dibanding versi yang muat di memori karena disk jauh lebih lambat.

Pengaturan yang Relevan #

Konfigurasi	Peran	Efek salah atur
`spark.driver.memory` / `spark.executor.memory`	ukuran heap JVM	terlalu kecil: spill atau OOM; terlalu besar: executor sedikit dan mahal
`spark.memory.fraction`	porsi heap untuk pool unified	mengecilkan memori kerja shuffle/sort/agregasi
`spark.memory.storageFraction`	batas lunak wilayah storage	terlalu besar: cache menahan ruang execution
`StorageLevel` pada `persist()`	format penyimpanan cache	`MEMORY_ONLY` pada DataFrame besar memicu rekomputasi saat blok di-evict

Dua catatan lanjutan: data cache dan shuffle berformat serialized ter-encode (hemat ruang, butuh biaya encode saat dibaca), dan untuk data yang melebihi RAM sekalipun, off-heap memory (spark.memory.offHeap.enabled) menyediakan wilayah tambahan yang tidak ikut beban pemantauan objek JVM. Seluruh kunci konfigurasi ini terdokumentasi di halaman konfigurasi Spark .

Konsumen execution memory terbesar dalam pekerjaan sehari-hari justru bukan orderBy — melainkan shuffle: semua Exchange yang kita lihat di execution plan. Bagaimana shuffle bekerja, mengapa satu kunci bisa menumpuk 88 ribu baris di satu task, dan bagaimana salting memecahnya, menjadi pokok bahasan chapter Shuffle dan Skew.