Dua pipeline Spark dengan logika bisnis yang sama bisa berbeda durasi sepuluh kali lipat hanya karena perbedaan cara menulis kode dan mengatur konfigurasi. Perbedaan sebesar itu tidak akan kita temukan dengan menebak — kita butuh angka dari eksekusi nyata. Pada chapter ini kita membaca execution plan lewat explain(), mengukur seberapa cepat cache() membayar kembali biayanya, lalu membandingkan UDF Python melawan fungsi built-in pada jutaan email pelanggan TokoKita. Seluruh angka berasal dari eksekusi Spark 4.2.0 di mesin lokal satu core yang sama dengan chapter sebelumnya.

Mulai dari Execution Plan #

Langkah optimasi apa pun sebaiknya dimulai dari explain(): metode ini mencetak physical plan — peta urutan operasi yang benar-benar dijalankan Spark. Plan dibaca dari baris paling bawah (sumber data) ke atas (hasil akhir), dan setiap baris bertingkat adalah satu tahap. Dua pola yang paling sering menjadi biang kerok durasi adalah Exchange (data berpindah antar partition lewat shuffle) dan Sort yang mendahului SortMergeJoin. Kita sudah pernah melihat Exchange pada chapter Spark Partitioning; sekarang kita membacanya dalam konteks pipeline lengkap.

Skrip Pengukuran dan Hasilnya #

Skrip berikut menyiapkan 20 juta pesanan tersebar di 8 cabang, meringkasnya per kota, lalu melakukan tiga pengukuran: explain() pada pipeline join-agregasi, efek cache() pada agregasi yang dihitung tiga kali, dan perlombaan UDF Python melawan fungsi built-in F.upper():

python 
"""Pengukuran performa Spark: execution plan, cache, dan UDF."""
import time

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

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

# Sumber data sintetis TokoKita: 20 juta pesanan tersebar di 8 cabang
pesanan = spark.range(20_000_000).select(
    (F.col("id") % 8).alias("cabang"),
    ((F.col("id") % 500) * 1000 + 15000).cast("double").alias("nilai"),
)

dim_cabang = spark.createDataFrame(
    [(0, "Jakarta"), (1, "Bandung"), (2, "Surabaya"), (3, "Medan"),
     (4, "Makassar"), (5, "Denpasar"), (6, "Semarang"), (7, "Yogyakarta")],
    "cabang INT, kota STRING",
)

print("=== Bagian 1: execution plan (explain) ===")
analisis = (pesanan.join(dim_cabang, "cabang")
            .groupBy("kota")
            .agg(F.count("*").alias("jumlah_pesanan"),
                 F.round(F.avg("nilai"), 1).alias("rata_nilai")))
analisis.explain()

print("=== Bagian 2: efek cache() pada agregasi berulang ===")
ringkasan = pesanan.groupBy("cabang").agg(
    F.count("*").alias("jumlah"),
    F.round(F.sum("nilai"), 0).alias("total_nilai"),
)
ringkasan.cache()

t0 = time.perf_counter()
n1 = ringkasan.count()
t1 = time.perf_counter()
print(f"count() pertama (mengisi cache) : {n1} baris, {t1 - t0:.2f} detik")

t0 = time.perf_counter()
n2 = ringkasan.count()
t1 = time.perf_counter()
print(f"count() kedua (dari cache)      : {n2} baris, {t1 - t0:.2f} detik")

ringkasan.unpersist()
t0 = time.perf_counter()
n3 = ringkasan.count()
t1 = time.perf_counter()
print(f"count() ketiga (cache dibuang)  : {n3} baris, {t1 - t0:.2f} detik")

print("=== Bagian 3: UDF Python vs fungsi built-in ===")
pengguna = spark.range(5_000_000).select(
    F.concat(F.lit("user-"), F.col("id").cast("string"),
             F.lit("@example.com")).alias("email"))
pengguna.cache()
pengguna.count()  # materialisasi ke cache agar yang terukur murni biaya fungsi

@F.udf(StringType())
def normalisasi_udf(teks):
    return teks.upper()

t0 = time.perf_counter()
hasil_bawaan = (pengguna.select(F.upper("email").alias("email_besar"))
                .agg(F.min("email_besar"), F.max("email_besar")).first())
t1 = time.perf_counter()
print(f"built-in F.upper : {t1 - t0:.2f} detik, contoh: {hasil_bawaan[1]}")

t0 = time.perf_counter()
hasil_udf = (pengguna.select(normalisasi_udf("email").alias("email_besar"))
             .agg(F.min("email_besar"), F.max("email_besar")).first())
t1 = time.perf_counter()
print(f"UDF Python       : {t1 - t0:.2f} detik, contoh: {hasil_udf[1]}")

spark.stop()

Output yang Dihasilkan:

text 
=== Bagian 1: execution plan (explain) ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[kota#4], functions=[count(1), avg(nilai#2)])
   +- Exchange hashpartitioning(kota#4, 2), ENSURE_REQUIREMENTS, [plan_id=44]
      +- HashAggregate(keys=[kota#4], functions=[partial_count(1), partial_avg(nilai#2)])
         +- Project [nilai#2, kota#4]
            +- SortMergeJoin [cabang#1L], [cast(cabang#3 as bigint)], Inner
               :- Sort [cabang#1L ASC NULLS FIRST], false, 0
               :  +- Exchange hashpartitioning(cabang#1L, 2), ENSURE_REQUIREMENTS, [plan_id=37]
               :     +- Project [(id#0L % 8) AS cabang#1L, cast((((id#0L % 500) * 1000) + 15000) as double) AS nilai#2]
               :        +- Filter isnotnull((id#0L % 8))
               :           +- Range (0, 20000000, step=1, splits=1)
               +- Sort [cast(cabang#3 as bigint) ASC NULLS FIRST], false, 0
                  +- Exchange hashpartitioning(cast(cabang#3 as bigint), 2), ENSURE_REQUIREMENTS, [plan_id=36]
                     +- Filter isnotnull(cabang#3)
                        +- Scan ExistingRDD[cabang#3,kota#4]


=== Bagian 2: efek cache() pada agregasi berulang ===
count() pertama (mengisi cache) : 8 baris, 2.17 detik
count() kedua (dari cache)      : 8 baris, 0.15 detik
count() ketiga (cache dibuang)  : 8 baris, 0.64 detik
=== Bagian 3: UDF Python vs fungsi built-in ===
built-in F.upper : 3.44 detik, contoh: [email protected]
UDF Python       : 6.66 detik, contoh: [email protected]

Mari kita telaah baris-baris kode di atas:

  1. Baris 11 (spark.sql.shuffle.partitions=2): menurunkan jumlah partition hasil shuffle dari default 200 menjadi 2 — angka yang sepadan dengan hasil agregasi 8 baris di mesin satu core.
  2. Baris 32 (analisis.explain()): mencetak physical plan tanpa mengeksekusi query — bukti sekali lagi bahwa Spark bersifat lazy.
  3. Baris 39 (ringkasan.cache()): menandai DataFrame untuk disimpan di memori saat dihitung pertama kali; pemanggilan ini sendiri tidak memicu komputasi.
  4. Baris 41-55 (tiga pengukuran count()): pengukuran pertama mengisi cache, pengukuran kedua membaca dari cache, dan unpersist() di baris 51 membuang cache sebelum pengukuran ketiga.
  5. Baris 62 (pengguna.count()): memaksa materialisasi DataFrame email ke cache lebih dulu, sehingga pengukuran bagian 3 murni mengukur biaya fungsi upper, bukan biaya membangun data.
  6. Baris 64-66 (@F.udf): mendeklarasikan UDF Python biasa — kode Python yang akan dieksekusi baris demi baris di proses worker Python.

Membaca Plan dan Efek cache() #

Plan bagian 1 memperlihatkan dua Exchange: satu untuk menyebar pesanan menurut cabang, satu lagi untuk menyebar hasil join menurut kota sebelum agregasi final. Ditambah dua Sort untuk SortMergeJoin, pipeline sederhana ini sudah memuat empat operasi mahal — itulah mengapa menghitung ulang pipeline yang sama berkali-kali adalah pemborosan. Bagian 2 membuktikannya dengan angka: count() pertama membayar 2,17 detik untuk meringkas 20 juta baris, count() kedua hanya 0,15 detik karena membaca 8 baris hasil dari cache, dan setelah unpersist() biayanya kembali 0,64 detik. Angka waktu pasti bergerak antar eksekusi (run lain di mesin yang sama mencatat 2,40/0,18/0,82 detik), tetapi polanya konsisten: membaca dari cache sepuluh kali lebih cepat daripada menghitung ulang.

Biaya UDF Python #

Bagian 3 mengukur jebakan yang paling sering diam-diam membebani pipeline: UDF. Fungsi built-in F.upper() menyelesaikan normalisasi 5 juta email dalam 3,44 detik, sementara UDF Python dengan logika identik butuh 6,66 detik — hampir dua kali lebih lambat di mesin lokal ini. Penyebabnya adalah arsitektur eksekusi: fungsi built-in berjalan langsung di JVM atas data ter-encode, sedangkan UDF Python memaksa Spark men-serialisasi tiap baris, mengirimkannya ke proses worker Python, menunggu hasil, lalu men-serialisasinya kembali. Di cluster dengan data lebih besar, rasio ini cenderung memburuk. Aturan praktisnya: periksa dulu daftar fungsi di modul pyspark.sql.functions sebelum menulis UDF, dan bila UDF tidak terhindarkan, pertimbangkan pandas_udf yang memproses data per batch, bukan per baris. Rangkaian teknik optimasi ini terdokumentasi lengkap di tuning guide Spark .

AQE dan Konfigurasi Partition #

Satu pengungkit performa besar sudah kita buktikan pada chapter Spark Partitioning: Adaptive Query Execution (AQE), aktif secara default di Spark 4, memangkas 200 partition shuffle menjadi 1 pada agregasi kecil. Daftar periksa sederhana untuk setiap job: berapa nilai spark.sql.shuffle.partitions dibanding volume data (default 200 sering terlalu besar untuk data kecil), berapa banyak Exchange di plan, dan DataFrame mana yang dipakai berulang sehingga layak di-cache. Seluruh daftar konfigurasi spark.sql.* terdokumentasi di halaman konfigurasi Spark .

Mengukur dan menata eksekusi batch hanyalah separuh cerita performa. Separuh lainnya muncul ketika data tidak lagi datang sebagai berkas lengkap semalam, melainkan mengalir baris demi baris sepanjang hari — dunia yang kita masuki pada chapter Spark Structured Streaming.