In-Memory Computing dengan Apache Spark

Arsitektur Apache Spark (driver, executor, cluster manager), Resilient Distributed Datasets, transformasi lazy vs aksi, narrow vs wide dependency, DAG & stage, caching, serta Spark DataFrame dan Spark SQL dengan PySpark.

Notebook lab (pratinjau)

Unduh lalu jalankan di Jupyter, VS Code, atau Google Colab.

Unduh lab.ipynb

Lab Minggu 13 — In-Memory Computing dengan Apache Spark

Tujuan lab:

Memahami RDD: partisi, lazy evaluation, lineage, narrow vs wide, cache — lewat MiniRDD (Python murni).
Menjalankan kode PySpark sungguhan (RDD & DataFrame & Spark SQL).
Membandingkan agregasi pandas vs Spark DataFrame pada log transaksi.

Catatan instalasi. Bagian PySpark butuh pip install pyspark dan Java 17. Jika belum tersedia, sel-sel Spark otomatis dilewati (notebook tetap jalan). Alternatif termudah: buka notebook ini di Google Colab lalu jalankan !pip install pyspark.

[1]

import sys
sys.path.insert(0, "..")

import time
import zlib
import functools
from collections import Counter

import pandas as pd
import matplotlib.pyplot as plt

from common.datasets import dokumen_teks, log_transaksi

0. Menyiapkan SparkSession (jika tersedia)

[2]

ADA_SPARK = False
try:
    from pyspark.sql import SparkSession, functions as F
    spark = (SparkSession.builder.master("local[*]").appName("lab13")
             .config("spark.sql.shuffle.partitions", "4").getOrCreate())
    sc = spark.sparkContext
    ADA_SPARK = True
    print("Spark", spark.version, "siap. Spark UI:", sc.uiWebUrl)
except Exception as e:  # ImportError atau Java tidak ditemukan
    print("PySpark belum tersedia ->", type(e).__name__)
    print("Install: pip install pyspark  (+ Java 17), atau jalankan di Google Colab.")
    print("Bagian MiniRDD & pandas tetap bisa dijalankan.")

1. MiniRDD — RDD mini untuk melihat "jeroan" Spark

Versi ringkas di bawah ini sama idenya dengan yang kamu lengkapi di exercise.py. Setiap transformasi hanya mencatat fungsi; eksekusi terjadi saat aksi.

[3]

class MiniRDD:
    def __init__(self, sumber=None, induk=None, fungsi=None, nama="parallelize"):
        self._sumber, self._induk, self._fungsi, self.nama = sumber, induk, fungsi, nama

    @classmethod
    def parallelize(cls, data, n=2):
        data = list(data); q, r = divmod(len(data), n); parts, i = [], 0
        for p in range(n):
            uk = q + (p < r); parts.append(data[i:i + uk]); i += uk
        return cls(sumber=parts)

    def _hitung(self):
        if self._induk is None:
            return [list(p) for p in self._sumber]
        print(f"   ⚙ menjalankan {self.nama}")
        return self._fungsi(self._induk._hitung())

    def lineage(self):
        return [self.nama] if self._induk is None else self._induk.lineage() + [self.nama]

    # transformasi (lazy)
    def map(self, f):    return MiniRDD(induk=self, nama="map", fungsi=lambda P: [[f(x) for x in p] for p in P])
    def filter(self, f): return MiniRDD(induk=self, nama="filter", fungsi=lambda P: [[x for x in p if f(x)] for p in P])
    def flatMap(self, f): return MiniRDD(induk=self, nama="flatMap", fungsi=lambda P: [[y for x in p for y in f(x)] for p in P])

    def reduceByKey(self, f):
        def jalan(P):
            n = len(P); tujuan = [{} for _ in range(n)]
            for p in P:
                lokal = {}
                for k, v in p: lokal[k] = f(lokal[k], v) if k in lokal else v
                for k, v in lokal.items():
                    d = tujuan[zlib.crc32(repr(k).encode()) % n]
                    d[k] = f(d[k], v) if k in d else v
            return [list(d.items()) for d in tujuan]
        return MiniRDD(induk=self, nama="reduceByKey [SHUFFLE]", fungsi=jalan)

    # aksi
    def glom(self):    return self._hitung()
    def collect(self): return [x for p in self._hitung() for x in p]
    def count(self):   return sum(len(p) for p in self._hitung())

1a. Lazy evaluation

[4]

rdd = MiniRDD.parallelize([1, 2, 3, 4, 5, 6], 3)
kuadrat = rdd.map(lambda x: x * x)
genap = kuadrat.filter(lambda x: x % 2 == 0)
print("Belum ada yang dijalankan. Lineage:", genap.lineage())

[5]

print("Aksi collect():")
print(genap.collect())
print("Per partisi:", genap.glom())

Panggil aksi lagi → lineage dihitung ulang (karena tidak di-cache):

[6]

print(genap.count())

1b. WordCount & shuffle

[7]

docs = dokumen_teks()
wc = (MiniRDD.parallelize(docs, 3)
      .flatMap(lambda d: d.split())
      .map(lambda w: (w, 1))
      .reduceByKey(lambda a, b: a + b))
print("Lineage:", " → ".join(wc.lineage()))
partisi = wc.glom()
for i, p in enumerate(partisi):
    print(f"partisi {i}: {len(p)} kata unik, contoh {p[:3]}")

Setelah reduceByKey, setiap kata hanya ada di satu partisi (ditentukan hash(key) % n). Lineage dengan shuffle → Spark memotongnya menjadi 2 stage: [parallelize, flatMap, map] | [reduceByKey].

2. PySpark RDD (sungguhan)

[8]

if ADA_SPARK:
    rdd = sc.parallelize([1, 2, 3, 4, 5, 6], 3)
    genap = rdd.map(lambda x: x * x).filter(lambda x: x % 2 == 0)
    print(genap.collect(), genap.glom().collect())
    print(genap.toDebugString().decode())          # lineage versi Spark

    wc = (sc.parallelize(docs, 3)
            .flatMap(lambda d: d.split())
            .map(lambda w: (w, 1))
            .reduceByKey(lambda a, b: a + b))
    print(sorted(wc.collect(), key=lambda kv: -kv[1])[:5])
else:
    print("(dilewati — PySpark belum tersedia)")

Cache: efeknya pada aksi berulang

[9]

if ADA_SPARK:
    besar = sc.parallelize(range(2_000_000), 8).map(lambda x: (x % 1000, x * 0.5))
    t = time.time(); besar.count(); besar.count(); print("tanpa cache:", round(time.time() - t, 2), "s")
    besar.cache(); besar.count()                       # materialisasi cache
    t = time.time(); besar.count(); besar.count(); print("dengan cache:", round(time.time() - t, 2), "s")
    besar.unpersist()
else:
    print("(dilewati — PySpark belum tersedia)")

3. DataFrame: pandas vs Spark

3a. pandas (satu mesin, eager)

[10]

log = log_transaksi()
pdf = pd.DataFrame([b.split(",") for b in log], columns=["tanggal", "kota", "produk", "jumlah", "harga"])
pdf["tanggal"] = pd.to_datetime(pdf["tanggal"])
pdf["jumlah"] = pdf["jumlah"].astype(int)
pdf["harga"] = pdf["harga"].astype(int)
pdf["pendapatan"] = pdf["jumlah"] * pdf["harga"]
pdf.head()

[11]

agg = (pdf.groupby("kota")
          .agg(total=("pendapatan", "sum"), n_transaksi=("pendapatan", "size"))
          .sort_values("total", ascending=False))
agg

[12]

pivot = pdf.pivot_table(index=pdf["tanggal"].dt.month.rename("bulan"), columns="kota",
                        values="pendapatan", aggfunc="sum", fill_value=0)
(pivot / 1e6).plot(figsize=(10, 4), marker="o", title="Pendapatan bulanan per kota (juta Rp)")
plt.show()

3b. Spark DataFrame (terdistribusi, lazy)

[13]

if ADA_SPARK:
    sdf = (spark.createDataFrame(pdf.astype({"tanggal": str}))
                .withColumn("tanggal", F.to_date("tanggal")))
    sdf.printSchema()

    hasil = (sdf.groupBy("kota")
                .agg(F.sum("pendapatan").alias("total"), F.count("*").alias("n_transaksi"))
                .orderBy(F.desc("total")))
    hasil.explain()          # rencana eksekusi dari Catalyst
    hasil.show()
else:
    print("(dilewati — PySpark belum tersedia)")

3c. Spark SQL

[14]

if ADA_SPARK:
    sdf.createOrReplaceTempView("log")
    spark.sql("""
        SELECT produk, SUM(pendapatan) AS total, ROUND(AVG(jumlah), 2) AS rata_qty
        FROM log
        GROUP BY produk
        ORDER BY total DESC
        LIMIT 5
    """).show()

    (sdf.withColumn("bulan", F.month("tanggal"))
        .groupBy("bulan").pivot("kota").sum("pendapatan")
        .orderBy("bulan").show())
else:
    print("(dilewati — PySpark belum tersedia)")

3d. Broadcast join: tabel dimensi kecil × tabel fakta besar

[15]

dim_kota = pd.DataFrame({"kota": ["Jakarta", "Bandung", "Surabaya", "Medan", "Makassar", "Bekasi"],
                         "pulau": ["Jawa", "Jawa", "Jawa", "Sumatra", "Sulawesi", "Jawa"]})
print(pdf.merge(dim_kota, on="kota").groupby("pulau")["pendapatan"].sum())

if ADA_SPARK:
    sdim = spark.createDataFrame(dim_kota)
    (sdf.join(F.broadcast(sdim), "kota")
        .groupBy("pulau").agg(F.sum("pendapatan").alias("total")).show())

✍️ Coba sendiri

Dengan MiniRDD, hitung jumlah transaksi per produk dari log (parse baris → (produk, 1) → reduceByKey).
Gambar DAG (tulis dengan teks) untuk pipeline nomor 1 dan tentukan jumlah stage.
Tulis padanan PySpark untuk: 5 kombinasi (kota, produk) dengan rata-rata qty tertinggi. Jalankan versi pandas-nya.
(Jika Spark tersedia) buka Spark UI (sc.uiWebUrl) dan catat jumlah stage & task dari job hasil.show().

[16]

# 1.

[17]

# 3.

[18]

if ADA_SPARK:
    spark.stop()

Selanjutnya

Kerjakan kuis dan exercise.py, lalu jalankan python check.py 13 dari folder utama.