{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "83d876d3",
   "metadata": {},
   "source": [
    "# Lab Minggu 13 — In-Memory Computing dengan Apache Spark\n",
    "\n",
    "**Tujuan lab:**\n",
    "1. Memahami RDD: partisi, lazy evaluation, lineage, narrow vs wide, cache — lewat **MiniRDD** (Python murni).\n",
    "2. Menjalankan kode **PySpark** sungguhan (RDD & DataFrame & Spark SQL).\n",
    "3. Membandingkan agregasi pandas vs Spark DataFrame pada log transaksi.\n",
    "\n",
    "> **Catatan instalasi.** Bagian PySpark butuh `pip install pyspark` **dan Java 17**.\n",
    "> Jika belum tersedia, sel-sel Spark otomatis dilewati (notebook tetap jalan).\n",
    "> Alternatif termudah: buka notebook ini di **Google Colab** lalu jalankan `!pip install pyspark`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be24c308",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.insert(0, \"..\")\n",
    "\n",
    "import time\n",
    "import zlib\n",
    "import functools\n",
    "from collections import Counter\n",
    "\n",
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "from common.datasets import dokumen_teks, log_transaksi"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1c0ff670",
   "metadata": {},
   "source": [
    "## 0. Menyiapkan SparkSession (jika tersedia)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "783838f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "ADA_SPARK = False\n",
    "try:\n",
    "    from pyspark.sql import SparkSession, functions as F\n",
    "    spark = (SparkSession.builder.master(\"local[*]\").appName(\"lab13\")\n",
    "             .config(\"spark.sql.shuffle.partitions\", \"4\").getOrCreate())\n",
    "    sc = spark.sparkContext\n",
    "    ADA_SPARK = True\n",
    "    print(\"Spark\", spark.version, \"siap. Spark UI:\", sc.uiWebUrl)\n",
    "except Exception as e:  # ImportError atau Java tidak ditemukan\n",
    "    print(\"PySpark belum tersedia ->\", type(e).__name__)\n",
    "    print(\"Install: pip install pyspark  (+ Java 17), atau jalankan di Google Colab.\")\n",
    "    print(\"Bagian MiniRDD & pandas tetap bisa dijalankan.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d1968a3e",
   "metadata": {},
   "source": [
    "## 1. MiniRDD — RDD mini untuk melihat \"jeroan\" Spark\n",
    "\n",
    "Versi ringkas di bawah ini sama idenya dengan yang kamu lengkapi di `exercise.py`.\n",
    "Setiap transformasi hanya **mencatat** fungsi; eksekusi terjadi saat aksi."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5fd33a81",
   "metadata": {},
   "outputs": [],
   "source": [
    "class MiniRDD:\n",
    "    def __init__(self, sumber=None, induk=None, fungsi=None, nama=\"parallelize\"):\n",
    "        self._sumber, self._induk, self._fungsi, self.nama = sumber, induk, fungsi, nama\n",
    "\n",
    "    @classmethod\n",
    "    def parallelize(cls, data, n=2):\n",
    "        data = list(data); q, r = divmod(len(data), n); parts, i = [], 0\n",
    "        for p in range(n):\n",
    "            uk = q + (p < r); parts.append(data[i:i + uk]); i += uk\n",
    "        return cls(sumber=parts)\n",
    "\n",
    "    def _hitung(self):\n",
    "        if self._induk is None:\n",
    "            return [list(p) for p in self._sumber]\n",
    "        print(f\"   ⚙ menjalankan {self.nama}\")\n",
    "        return self._fungsi(self._induk._hitung())\n",
    "\n",
    "    def lineage(self):\n",
    "        return [self.nama] if self._induk is None else self._induk.lineage() + [self.nama]\n",
    "\n",
    "    # transformasi (lazy)\n",
    "    def map(self, f):    return MiniRDD(induk=self, nama=\"map\", fungsi=lambda P: [[f(x) for x in p] for p in P])\n",
    "    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])\n",
    "    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])\n",
    "\n",
    "    def reduceByKey(self, f):\n",
    "        def jalan(P):\n",
    "            n = len(P); tujuan = [{} for _ in range(n)]\n",
    "            for p in P:\n",
    "                lokal = {}\n",
    "                for k, v in p: lokal[k] = f(lokal[k], v) if k in lokal else v\n",
    "                for k, v in lokal.items():\n",
    "                    d = tujuan[zlib.crc32(repr(k).encode()) % n]\n",
    "                    d[k] = f(d[k], v) if k in d else v\n",
    "            return [list(d.items()) for d in tujuan]\n",
    "        return MiniRDD(induk=self, nama=\"reduceByKey [SHUFFLE]\", fungsi=jalan)\n",
    "\n",
    "    # aksi\n",
    "    def glom(self):    return self._hitung()\n",
    "    def collect(self): return [x for p in self._hitung() for x in p]\n",
    "    def count(self):   return sum(len(p) for p in self._hitung())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5aa5e4ab",
   "metadata": {},
   "source": [
    "### 1a. Lazy evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "04d5d684",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd = MiniRDD.parallelize([1, 2, 3, 4, 5, 6], 3)\n",
    "kuadrat = rdd.map(lambda x: x * x)\n",
    "genap = kuadrat.filter(lambda x: x % 2 == 0)\n",
    "print(\"Belum ada yang dijalankan. Lineage:\", genap.lineage())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8e52af66",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Aksi collect():\")\n",
    "print(genap.collect())\n",
    "print(\"Per partisi:\", genap.glom())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae2d27d4",
   "metadata": {},
   "source": [
    "Panggil aksi lagi → lineage dihitung **ulang** (karena tidak di-cache):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "530ae50a",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(genap.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "173d3447",
   "metadata": {},
   "source": [
    "### 1b. WordCount & shuffle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b9b28781",
   "metadata": {},
   "outputs": [],
   "source": [
    "docs = dokumen_teks()\n",
    "wc = (MiniRDD.parallelize(docs, 3)\n",
    "      .flatMap(lambda d: d.split())\n",
    "      .map(lambda w: (w, 1))\n",
    "      .reduceByKey(lambda a, b: a + b))\n",
    "print(\"Lineage:\", \" → \".join(wc.lineage()))\n",
    "partisi = wc.glom()\n",
    "for i, p in enumerate(partisi):\n",
    "    print(f\"partisi {i}: {len(p)} kata unik, contoh {p[:3]}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e3a81fc",
   "metadata": {},
   "source": [
    "Setelah `reduceByKey`, setiap kata hanya ada di **satu** partisi (ditentukan `hash(key) % n`).\n",
    "Lineage dengan shuffle → Spark memotongnya menjadi **2 stage**:\n",
    "`[parallelize, flatMap, map]` | `[reduceByKey]`."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "89686119",
   "metadata": {},
   "source": [
    "## 2. PySpark RDD (sungguhan)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "08439547",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    rdd = sc.parallelize([1, 2, 3, 4, 5, 6], 3)\n",
    "    genap = rdd.map(lambda x: x * x).filter(lambda x: x % 2 == 0)\n",
    "    print(genap.collect(), genap.glom().collect())\n",
    "    print(genap.toDebugString().decode())          # lineage versi Spark\n",
    "\n",
    "    wc = (sc.parallelize(docs, 3)\n",
    "            .flatMap(lambda d: d.split())\n",
    "            .map(lambda w: (w, 1))\n",
    "            .reduceByKey(lambda a, b: a + b))\n",
    "    print(sorted(wc.collect(), key=lambda kv: -kv[1])[:5])\n",
    "else:\n",
    "    print(\"(dilewati — PySpark belum tersedia)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fe7427b7",
   "metadata": {},
   "source": [
    "### Cache: efeknya pada aksi berulang"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f96c0b70",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    besar = sc.parallelize(range(2_000_000), 8).map(lambda x: (x % 1000, x * 0.5))\n",
    "    t = time.time(); besar.count(); besar.count(); print(\"tanpa cache:\", round(time.time() - t, 2), \"s\")\n",
    "    besar.cache(); besar.count()                       # materialisasi cache\n",
    "    t = time.time(); besar.count(); besar.count(); print(\"dengan cache:\", round(time.time() - t, 2), \"s\")\n",
    "    besar.unpersist()\n",
    "else:\n",
    "    print(\"(dilewati — PySpark belum tersedia)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "330b1a61",
   "metadata": {},
   "source": [
    "## 3. DataFrame: pandas vs Spark\n",
    "\n",
    "### 3a. pandas (satu mesin, eager)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b415c361",
   "metadata": {},
   "outputs": [],
   "source": [
    "log = log_transaksi()\n",
    "pdf = pd.DataFrame([b.split(\",\") for b in log], columns=[\"tanggal\", \"kota\", \"produk\", \"jumlah\", \"harga\"])\n",
    "pdf[\"tanggal\"] = pd.to_datetime(pdf[\"tanggal\"])\n",
    "pdf[\"jumlah\"] = pdf[\"jumlah\"].astype(int)\n",
    "pdf[\"harga\"] = pdf[\"harga\"].astype(int)\n",
    "pdf[\"pendapatan\"] = pdf[\"jumlah\"] * pdf[\"harga\"]\n",
    "pdf.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3435779c",
   "metadata": {},
   "outputs": [],
   "source": [
    "agg = (pdf.groupby(\"kota\")\n",
    "          .agg(total=(\"pendapatan\", \"sum\"), n_transaksi=(\"pendapatan\", \"size\"))\n",
    "          .sort_values(\"total\", ascending=False))\n",
    "agg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0ac1af3c",
   "metadata": {},
   "outputs": [],
   "source": [
    "pivot = pdf.pivot_table(index=pdf[\"tanggal\"].dt.month.rename(\"bulan\"), columns=\"kota\",\n",
    "                        values=\"pendapatan\", aggfunc=\"sum\", fill_value=0)\n",
    "(pivot / 1e6).plot(figsize=(10, 4), marker=\"o\", title=\"Pendapatan bulanan per kota (juta Rp)\")\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77efd560",
   "metadata": {},
   "source": [
    "### 3b. Spark DataFrame (terdistribusi, lazy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6783e286",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    sdf = (spark.createDataFrame(pdf.astype({\"tanggal\": str}))\n",
    "                .withColumn(\"tanggal\", F.to_date(\"tanggal\")))\n",
    "    sdf.printSchema()\n",
    "\n",
    "    hasil = (sdf.groupBy(\"kota\")\n",
    "                .agg(F.sum(\"pendapatan\").alias(\"total\"), F.count(\"*\").alias(\"n_transaksi\"))\n",
    "                .orderBy(F.desc(\"total\")))\n",
    "    hasil.explain()          # rencana eksekusi dari Catalyst\n",
    "    hasil.show()\n",
    "else:\n",
    "    print(\"(dilewati — PySpark belum tersedia)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "102e0f0d",
   "metadata": {},
   "source": [
    "### 3c. Spark SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "16202e16",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    sdf.createOrReplaceTempView(\"log\")\n",
    "    spark.sql(\"\"\"\n",
    "        SELECT produk, SUM(pendapatan) AS total, ROUND(AVG(jumlah), 2) AS rata_qty\n",
    "        FROM log\n",
    "        GROUP BY produk\n",
    "        ORDER BY total DESC\n",
    "        LIMIT 5\n",
    "    \"\"\").show()\n",
    "\n",
    "    (sdf.withColumn(\"bulan\", F.month(\"tanggal\"))\n",
    "        .groupBy(\"bulan\").pivot(\"kota\").sum(\"pendapatan\")\n",
    "        .orderBy(\"bulan\").show())\n",
    "else:\n",
    "    print(\"(dilewati — PySpark belum tersedia)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fa84f748",
   "metadata": {},
   "source": [
    "### 3d. Broadcast join: tabel dimensi kecil × tabel fakta besar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e359dfa",
   "metadata": {},
   "outputs": [],
   "source": [
    "dim_kota = pd.DataFrame({\"kota\": [\"Jakarta\", \"Bandung\", \"Surabaya\", \"Medan\", \"Makassar\", \"Bekasi\"],\n",
    "                         \"pulau\": [\"Jawa\", \"Jawa\", \"Jawa\", \"Sumatra\", \"Sulawesi\", \"Jawa\"]})\n",
    "print(pdf.merge(dim_kota, on=\"kota\").groupby(\"pulau\")[\"pendapatan\"].sum())\n",
    "\n",
    "if ADA_SPARK:\n",
    "    sdim = spark.createDataFrame(dim_kota)\n",
    "    (sdf.join(F.broadcast(sdim), \"kota\")\n",
    "        .groupBy(\"pulau\").agg(F.sum(\"pendapatan\").alias(\"total\")).show())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f682baf",
   "metadata": {},
   "source": [
    "## ✍️ Coba sendiri\n",
    "\n",
    "1. Dengan MiniRDD, hitung **jumlah transaksi per produk** dari `log` (parse baris → `(produk, 1)` → `reduceByKey`).\n",
    "2. Gambar DAG (tulis dengan teks) untuk pipeline nomor 1 dan tentukan jumlah stage.\n",
    "3. Tulis padanan PySpark untuk: *5 kombinasi (kota, produk) dengan rata-rata qty tertinggi*. Jalankan versi pandas-nya.\n",
    "4. (Jika Spark tersedia) buka Spark UI (`sc.uiWebUrl`) dan catat jumlah stage & task dari job `hasil.show()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e6e4984c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fba76c69",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b8a7424",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1cf80669",
   "metadata": {},
   "source": [
    "## Selanjutnya\n",
    "Kerjakan kuis dan `exercise.py`, lalu jalankan `python check.py 13` dari folder utama."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}