In-Memory Computing dengan Apache Spark

Capaian Pembelajaran

Setelah minggu ini kamu bisa:

1. Menjelaskan mengapa Spark lebih cepat dari MapReduce untuk beban iteratif & interaktif.
2. Menjelaskan arsitektur Spark: driver, cluster manager, executor, task.
3. Menjelaskan sifat RDD (Resilient Distributed Dataset): partisi, immutable, lineage, lazy.
4. Membedakan transformasi vs aksi, serta dependensi narrow vs wide (shuffle).
5. Membaca DAG → stage → task, dan memakai cache()/persist() dengan tepat.
6. Mengolah data dengan Spark DataFrame dan Spark SQL (PySpark).

1. Dari MapReduce ke Spark

MapReduce (iterasi K-Means 3x):
  HDFS ─► Map ─► Reduce ─► HDFS ─► Map ─► Reduce ─► HDFS ─► Map ─► Reduce ─► HDFS
          (baca disk)    (tulis disk)  (baca disk) ...

Spark:
  HDFS ─► [RDD di memori] ─► iterasi 1 ─► iterasi 2 ─► iterasi 3 ─► hasil
                 ▲ dibaca sekali, dipakai berulang dari RAM

Spark (UC Berkeley AMPLab, 2009; Apache 2013) tidak menggantikan HDFS — Spark sering membaca data dari HDFS/S3 dan berjalan di atas YARN atau Kubernetes.

2. Arsitektur Spark

         ┌──────────────────────────┐
         │  Driver Program          │  - menjalankan main() / notebook kamu
         │  SparkSession/Context    │  - membangun DAG, menjadwalkan task
         └────────────┬─────────────┘
                      │ minta sumber daya
         ┌────────────▼─────────────┐
         │  Cluster Manager         │  Standalone / YARN / Kubernetes
         └────┬───────────────┬─────┘
              │               │
     ┌────────▼──────┐ ┌──────▼────────┐
     │ Worker Node 1 │ │ Worker Node 2 │
     │ ┌───────────┐ │ │ ┌───────────┐ │
     │ │ Executor  │ │ │ │ Executor  │ │   executor = proses JVM yang
     │ │ task task │ │ │ │ task task │ │   menjalankan task & menyimpan
     │ │ [cache]   │ │ │ │ [cache]   │ │   partisi yang di-cache
     │ └───────────┘ │ │ └───────────┘ │
     └───────────────┘ └───────────────┘

• Task = unit kerja terkecil: satu task memproses satu partisi.
• Mode local[*] = driver & executor di satu mesin dengan semua core — cocok untuk belajar.

3. RDD — Resilient Distributed Dataset

RDD = koleksi elemen yang:

rdd = sc.parallelize([1, 2, 3, 4, 5, 6], numSlices=3)   # 3 partisi: [1,2] [3,4] [5,6]
kuadrat = rdd.map(lambda x: x * x)        # transformasi: BELUM dihitung
genap   = kuadrat.filter(lambda x: x % 2 == 0)
genap.collect()                            # aksi: SEKARANG dihitung → [4, 16, 36]

Lineage genap: parallelize → map → filter. Jika partisi ke-2 hilang, Spark cukup menghitung ulang [3,4] → [9,16] → [16].

3a. Transformasi vs Aksi

Mengapa lazy? Spark melihat seluruh rantai sebelum menjalankan, sehingga bisa mengoptimasi: menggabungkan beberapa map/filter dalam satu lintasan (pipelining) dan tidak menghitung yang tidak dipakai.

3b. Narrow vs Wide dependency

 NARROW (tanpa shuffle)               WIDE (butuh shuffle)
 map / filter / flatMap               reduceByKey / groupByKey / join / distinct

 P1 ──► P1'                           P1 ──┐──► P1'
 P2 ──► P2'                           P2 ──┼──► P2'
 P3 ──► P3'                           P3 ──┘──► P3'
 tiap partisi anak hanya              partisi anak butuh data dari
 butuh SATU partisi induk             BANYAK partisi induk (lewat jaringan)

• Stage = rangkaian transformasi narrow yang di-pipeline. Setiap wide dependency memotong stage baru.
• Jumlah task dalam stage = jumlah partisi.

Contoh: textFile → flatMap → map → reduceByKey → filter → collect

Stage 1: textFile → flatMap → map  (narrow, di-pipeline)  ── shuffle ──►
Stage 2: reduceByKey → filter → collect

→ 1 shuffle, 2 stage. Jika input 4 partisi dan hasil reduceByKey juga 4 partisi → 4 + 4 = 8 task.

3c. reduceByKey vs groupByKey

pasangan.reduceByKey(lambda a, b: a + b)     # ✔ menggabungkan lokal dulu (seperti combiner)
pasangan.groupByKey().mapValues(sum)         # ✘ semua nilai dikirim lewat jaringan

Hasilnya sama, tetapi reduceByKey jauh lebih hemat shuffle — konsep combiner dari Minggu 12.

3d. Cache / Persist

Tanpa cache, setiap aksi menghitung ulang seluruh lineage.

bersih = raw.map(parse).filter(valid)
bersih.cache()                 # = persist(StorageLevel.MEMORY_ONLY) untuk RDD
bersih.count()                 # dihitung + disimpan di memori executor
bersih.filter(...).count()     # memakai cache, tidak parse ulang
bersih.unpersist()

Cache ketika dataset dipakai berkali-kali (ML iteratif, eksplorasi interaktif). Jangan cache semuanya — memori terbatas.

4. Spark DataFrame & Spark SQL

DataFrame = data terdistribusi bertabel dengan skema (kolom bernama & bertipe), mirip tabel SQL / pandas DataFrame, tetapi terdistribusi & lazy.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[*]").appName("lab13").getOrCreate()

df = (spark.read.csv("log_transaksi.csv", header=False, inferSchema=True)
          .toDF("tanggal", "kota", "produk", "jumlah", "harga")
          .withColumn("pendapatan", F.col("jumlah") * F.col("harga"))
          .withColumn("bulan", F.month(F.to_date("tanggal"))))

(df.groupBy("kota")
   .agg(F.sum("pendapatan").alias("total"), F.count("*").alias("n_transaksi"))
   .orderBy(F.desc("total"))
   .show())

df.createOrReplaceTempView("log")
spark.sql("""
    SELECT produk, SUM(pendapatan) AS total
    FROM log GROUP BY produk ORDER BY total DESC LIMIT 5
""").show()

Catalyst mengubah kueri: logical plan → optimasi (predicate pushdown, column pruning, konstanta) → physical plan. Lihat dengan df.explain().

Padanan pandas ↔ PySpark:

Perbedaan penting: pandas eager & satu mesin; Spark lazy & terdistribusi — show(), count(), collect(), toPandas() adalah aksi. Hati-hati collect()/toPandas() pada data besar: semua data ditarik ke driver → out of memory.

5. Partisi & Shuffle dalam Praktik

• rdd.getNumPartitions() / df.rdd.getNumPartitions()
• repartition(n) → shuffle penuh, bisa menambah/mengurangi partisi.
• coalesce(n) → mengurangi partisi tanpa shuffle penuh.
• spark.sql.shuffle.partitions (default 200) → jumlah partisi setelah shuffle DataFrame.
• Aturan praktis: partisi ±100–200 MB, jumlah partisi ≥ 2–3× total core.
• Data skew: satu key sangat dominan → satu task jauh lebih lama. Solusi: salting, broadcast join.

Broadcast join: tabel kecil (dimensi) dikirim utuh ke semua executor sehingga tabel besar tidak perlu di-shuffle: big.join(F.broadcast(small), "id").

6. Spark UI

Saat SparkSession berjalan, buka http://localhost:4040: lihat Jobs → Stages → Tasks, durasi, ukuran shuffle, dan Storage (data yang di-cache). Alat utama untuk mendiagnosis job lambat.

7. Menyiapkan PySpark

pip install pyspark            # butuh Java 17 (atau 11/21 sesuai versi Spark)
java -version                  # pastikan Java terpasang & JAVA_HOME diset

Alternatif tanpa instalasi: Google Colab (!pip install pyspark, Java sudah tersedia). Di lab ini, bagian PySpark otomatis dilewati jika Spark belum tersedia, dan konsepnya dipraktikkan dengan MiniRDD — implementasi RDD mini di Python murni.

Ringkasan

• Spark menyimpan data antara di memori → cepat untuk iteratif & interaktif.
• RDD: terpartisi, immutable, lazy, pulih lewat lineage.
• Transformasi (lazy) vs aksi (eksekusi). Wide dependency → shuffle → stage baru.
• reduceByKey > groupByKey; cache() bila data dipakai berulang.
• Gunakan DataFrame/Spark SQL untuk data tabel: dioptimasi Catalyst & Tungsten.

Bacaan

• Zaharia dkk. (2012) — Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing.
• Chambers & Zaharia — Spark: The Definitive Guide.
• Dokumentasi resmi: RDD Programming Guide dan Spark SQL, DataFrames and Datasets Guide.

Tugas Minggu Ini

1. Kerjakan lab.ipynb sampai selesai.
2. Kerjakan kuis (di web, atau baca quiz.json).
3. Isi exercise.py, lalu cek: python check.py 13