Bagian 2: Menyiapkan Spark

SparkSession adalah pintu masuk seluruh pekerjaan: setiap demo, pipeline, dan pengujian pada proyek ini memanggil fungsi yang sama agar konfigurasinya tidak pernah berbeda antar bagian. Mesin uji kita hanya punya dua CPU dan sekitar 3,9 GB RAM, jadi konfigurasi bawaan yang dirancang untuk cluster — misalnya 200 shuffle partition — justru menambah beban. Pada bagian ini kita menyusun satu berkas YAML sebagai sumber kebenaran, satu fungsi pembuat session, lalu memverifikasi nilai aktifnya pada session yang benar-benar berjalan. Materi pendukung ada pada chapter Apache Spark; prinsip “cocok untuk mesin kecil, bukan mesin sebesar apa pun” dibahas pada Kapan Menggunakan Spark.

Satu Sumber Kebenaran: config/spark-local.yaml #

Seluruh konfigurasi proyek hidup pada satu berkas config/spark-local.yaml. Fungsi muat_konfigurasi pada src/session.py membaca berkas tersebut dan menyerapkannya ke nilai bawaan; bila berkas hilang, proyek tetap berjalan dengan nilai bawaan:

spark:
  master: "local[1]"
  aplikasi: "TokoKita-Pemrosesan"
  konfigurasi:
    # Satu inti CPU: cukup untuk dataset ~60 MB dan menjaga suite tetap ringan.
    spark.ui.enabled: false
    spark.ui.showConsoleProgress: false
    spark.log.level: "ERROR"
    # 200 (bawaan) dirancang untuk cluster; 8 lebih pas untuk local[1].
    spark.sql.shuffle.partitions: 8
    # Adaptive Query Execution aktif: Spark melebur shuffle partition kecil.
    spark.sql.adaptive.enabled: true
    # Ambang broadcast join: 10 MB (nilai bawaan Spark).
    spark.sql.autoBroadcastJoinThreshold: 10485760

Mari kita telaah arti setiap kunci pada tabel berikut:

Membuat SparkSession dari Konfigurasi #

Fungsi buat_session pada src/session.py (baris 55-91) menerapkan konfigurasi tersebut. Intinya:

    cfg = muat_konfigurasi()
    if nama_app is not None:
        cfg["spark.app.name"] = nama_app
    if shuffle_partitions is not None:
        cfg["spark.sql.shuffle.partitions"] = str(shuffle_partitions)
    if ekstra:
        cfg.update({k: _teks(v) for k, v in ekstra.items()})

    builder = (
        SparkSession.builder.master(cfg["spark.master"])
        .appName(cfg["spark.app.name"])
        .config("spark.ui.enabled", cfg["spark.ui.enabled"])
        .config("spark.ui.showConsoleProgress", cfg["spark.ui.showConsoleProgress"])
        .config("spark.log.level", cfg["spark.log.level"])
    )
    for kunci in (
        "spark.sql.shuffle.partitions",
        "spark.sql.adaptive.enabled",
        "spark.sql.autoBroadcastJoinThreshold",
    ):
        builder = builder.config(kunci, cfg[kunci])
    if ekstra:
        for kunci, nilai in ekstra.items():
            builder = builder.config(kunci, _teks(nilai))
    return builder.getOrCreate()

Mari kita telaah baris-baris kode di atas:

1. Baris 67 (cfg = muat_konfigurasi()): dict datar dibaca dari YAML sekali; seluruh kunci berformat spark.<nama>.
2. Baris 68-73 (penimpaan): argumen nama_app, shuffle_partitions, dan ekstra menimpa nilai YAML — mekanisme ini dipakai pengujian (misalnya shuffle_partitions=8 pada tests/conftest.py) dan dipakai demo Bagian 5 untuk memaksa sort-merge join lewat ekstra.
3. Baris 75-81 (builder): rantai master, appName, lalu config per kunci; nilai bool dinormalisasi menjadi "true"/"false" oleh _teks.
4. Baris 82-87 (loop SQL): tiga kunci spark.sql.* diterapkan dari dict; menambah kunci baru cukup dengan mengubah YAML, tanpa menyentuh fungsi ini.
5. Baris 91 (getOrCreate()): bila session dengan konfigurasi yang sama sudah ada, session itu dipakai ulang; spark.stop() pada berhenti() melepaskan JVM setelah skrip selesai.

Verifikasi Nilai Aktif pada Session #

Deklarasi konfigurasi saja belum cukup; kita membaca kembali nilai aktif dari session yang berjalan. Skrip demo_p2_session.py (dijalankan pada mesin uji dua CPU) mencetak hasil berikut:

Output yang Dihasilkan:

== muat_konfigurasi() ==
  spark.master                             local[1]
  spark.app.name                           TokoKita-Pemrosesan
  spark.ui.enabled                         false
  spark.ui.showConsoleProgress             false
  spark.log.level                          ERROR
  spark.sql.shuffle.partitions             8
  spark.sql.adaptive.enabled               true
  spark.sql.autoBroadcastJoinThreshold     10485760
== nilai aktif pada session ==
  master                : local[1]
  defaultParallelism    : 1
  spark.version          : 4.2.0
  ui.enabled             : false
  shuffle.partitions     : 8
  adaptive.enabled       : true
  autoBroadcastJoinThreshold: 10485760
  log.level              : ERROR
  waktu pembuatan session: 6.79 s
== aksi pertama (pemanasan) ==
  spark.range(1_000_000).count() = 1,000,000 dalam 2.527 s

Tiga angka pada keluaran tersebut layak dicermati. Pertama, defaultParallelism bernilai 1 — konsekuensi langsung local[1] — sehingga operasi tanpa shuffle berjalan pada satu tugas. Kedua, pembuatan session pertama memakan 6,79 detik (JVM dan modul Spark dimuat); sesi berikutnya dalam proses yang sama akan jauh lebih cepat. Ketiga, aksi pertama setelah session hidup memakan 2,527 detik karena JIT dan pemuatan kelas masih berlangsung; angka-angka pada bagian-bagian berikutnya sudah menghiraukan efek pemanasan ini.

Disiplin Eksekusi pada Mesin Kecil #

Uji tests/test_session.py mengunci kontrak ini: nilai local[1], UI false, shuffle 8, AQE true, ambang broadcast 10485760, dan versi 4.2.0 diperiksa pada setiap eksekusi suite. Daftar lengkap kunci konfigurasi ada di https://spark.apache.org/docs/latest/configuration.html dan dokumentasi utama Spark di https://spark.apache.org/docs/latest/ .

Mesin sudah hidup dengan konfigurasi yang terkunci. Sekarang kita menghadapkan session tersebut ke dataset: bagaimana Spark membaca parquet terdistribusi, dan bagaimana struktur partition memengaruhi kecepatan baca — tugas Bagian 3.