{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "a7a693a6",
   "metadata": {},
   "source": [
    "# Lab Minggu 14 — Machine Learning Skala Besar (Spark MLlib)\n",
    "\n",
    "**Tujuan lab:**\n",
    "1. Melihat \"di balik layar\" MLlib: gradient descent & statistik **terdistribusi** dengan NumPy.\n",
    "2. Membuktikan hasil terdistribusi = hasil satu mesin.\n",
    "3. Membangun **Pipeline** Spark MLlib (klasifikasi churn & K-Means) — jika PySpark tersedia.\n",
    "4. Membandingkan dengan pipeline scikit-learn.\n",
    "\n",
    "> Bagian PySpark butuh `pip install pyspark` + Java 17 (atau Google Colab). Jika belum ada, sel Spark dilewati."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "db67f6f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.insert(0, \"..\")\n",
    "\n",
    "import time\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "from sklearn.cluster import KMeans\n",
    "from sklearn.datasets import make_blobs\n",
    "from sklearn.linear_model import LogisticRegression\n",
    "from sklearn.metrics import roc_auc_score, accuracy_score\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.pipeline import Pipeline\n",
    "from sklearn.preprocessing import StandardScaler, OneHotEncoder\n",
    "from sklearn.compose import ColumnTransformer\n",
    "\n",
    "from common.datasets import churn_pelanggan, segmentasi_pelanggan"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e09ec627",
   "metadata": {},
   "outputs": [],
   "source": [
    "ADA_SPARK = False\n",
    "try:\n",
    "    from pyspark.sql import SparkSession, functions as F\n",
    "    spark = (SparkSession.builder.master(\"local[*]\").appName(\"lab14\")\n",
    "             .config(\"spark.sql.shuffle.partitions\", \"4\").getOrCreate())\n",
    "    ADA_SPARK = True\n",
    "    print(\"Spark\", spark.version, \"siap\")\n",
    "except Exception as e:\n",
    "    print(\"PySpark belum tersedia ->\", type(e).__name__)\n",
    "    print(\"Install: pip install pyspark (+ Java 17), atau jalankan di Google Colab.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6a760b3c",
   "metadata": {},
   "source": [
    "## 1. Data churn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f19bcf00",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = churn_pelanggan()\n",
    "FITUR = [\"lama_langganan\", \"biaya_bulanan\", \"jumlah_keluhan\", \"jam_pemakaian\"]\n",
    "print(df.shape, \"| proporsi churn:\", df.churn.mean())\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "13a81c70",
   "metadata": {},
   "source": [
    "## 2. StandardScaler terdistribusi: (n, Σx, Σx²)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "727b8bbb",
   "metadata": {},
   "outputs": [],
   "source": [
    "X = df[FITUR].to_numpy(float)\n",
    "y = df[\"churn\"].to_numpy(float)\n",
    "partisi = np.array_split(X, 8)                     # bayangkan 8 executor\n",
    "\n",
    "stat = [(len(p), p.sum(0), (p ** 2).sum(0)) for p in partisi]      # map\n",
    "n = sum(s[0] for s in stat); S1 = sum(s[1] for s in stat); S2 = sum(s[2] for s in stat)  # reduce\n",
    "mean = S1 / n\n",
    "std = np.sqrt(S2 / n - mean ** 2)\n",
    "\n",
    "ref = StandardScaler().fit(X)\n",
    "pd.DataFrame({\"mean_terdistribusi\": mean, \"mean_sklearn\": ref.mean_,\n",
    "              \"std_terdistribusi\": std, \"std_sklearn\": ref.scale_}, index=FITUR)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "080a53c4",
   "metadata": {},
   "source": [
    "## 3. Regresi logistik: gradient descent terdistribusi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e7dedd2",
   "metadata": {},
   "outputs": [],
   "source": [
    "def sigmoid(z):\n",
    "    return 1 / (1 + np.exp(-np.clip(z, -500, 500)))\n",
    "\n",
    "Xs = (X - mean) / std\n",
    "Xb = np.column_stack([np.ones(len(Xs)), Xs])\n",
    "\n",
    "def latih(Xb, y, n_partisi=1, lr=0.5, epoch=300, catat=False):\n",
    "    parts = list(zip(np.array_split(Xb, n_partisi), np.array_split(y, n_partisi)))\n",
    "    w = np.zeros(Xb.shape[1]); riwayat = []\n",
    "    for _ in range(epoch):\n",
    "        hasil = [(Xp.T @ (sigmoid(Xp @ w) - yp), len(yp)) for Xp, yp in parts]  # map (di executor)\n",
    "        g = sum(h[0] for h in hasil); nn = sum(h[1] for h in hasil)            # treeAggregate\n",
    "        w = w - lr * g / nn                                                     # update di driver\n",
    "        if catat:\n",
    "            p = sigmoid(Xb @ w)\n",
    "            riwayat.append(-np.mean(y * np.log(p + 1e-12) + (1 - y) * np.log(1 - p + 1e-12)))\n",
    "    return w, riwayat\n",
    "\n",
    "w1, loss = latih(Xb, y, n_partisi=1, catat=True)\n",
    "w8, _ = latih(Xb, y, n_partisi=8)\n",
    "print(\"w (1 mesin)   :\", w1.round(4))\n",
    "print(\"w (8 partisi) :\", w8.round(4))\n",
    "print(\"Selisih maksimum:\", np.abs(w1 - w8).max())\n",
    "\n",
    "plt.plot(loss); plt.xlabel(\"epoch\"); plt.ylabel(\"log-loss\"); plt.title(\"Konvergensi GD\"); plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c659aac",
   "metadata": {},
   "outputs": [],
   "source": [
    "proba = sigmoid(Xb @ w8)\n",
    "print(f\"Akurasi: {accuracy_score(y, proba >= 0.5):.3f} | AUC: {roc_auc_score(y, proba):.3f}\")\n",
    "pd.Series(w8[1:], index=FITUR).plot.barh(title=\"Koefisien (fitur terstandarisasi)\"); plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cda54785",
   "metadata": {},
   "source": [
    "Interpretasi: koefisien negatif `lama_langganan` → pelanggan lama lebih jarang churn;\n",
    "koefisien positif `jumlah_keluhan` → banyak keluhan meningkatkan peluang churn.\n",
    "\n",
    "## 4. K-Means terdistribusi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28adc1eb",
   "metadata": {},
   "outputs": [],
   "source": [
    "Xk, _ = make_blobs(n_samples=900, centers=3, cluster_std=1.0, random_state=7)\n",
    "parts = np.array_split(Xk, 6)\n",
    "c = Xk[[0, 1, 2]].astype(float)\n",
    "jejak = [c.copy()]\n",
    "for it in range(15):\n",
    "    jumlah = np.zeros_like(c); banyak = np.zeros(len(c))\n",
    "    for p in parts:                                               # map per partisi\n",
    "        lab = np.linalg.norm(p[:, None] - c[None], axis=2).argmin(1)\n",
    "        np.add.at(jumlah, lab, p); banyak += np.bincount(lab, minlength=len(c))\n",
    "    baru = np.where(banyak[:, None] > 0, jumlah / np.maximum(banyak, 1)[:, None], c)  # driver\n",
    "    if np.abs(baru - c).max() < 1e-6:\n",
    "        break\n",
    "    c = baru; jejak.append(c.copy())\n",
    "print(\"Konvergen setelah\", it, \"iterasi\")\n",
    "\n",
    "jejak = np.array(jejak)\n",
    "plt.scatter(Xk[:, 0], Xk[:, 1], s=5, alpha=0.3)\n",
    "for j in range(3):\n",
    "    plt.plot(jejak[:, j, 0], jejak[:, j, 1], \"o-\", c=\"r\")\n",
    "plt.title(\"Pergerakan centroid (K-Means terdistribusi)\"); plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b807b96a",
   "metadata": {},
   "source": [
    "## 5. Pipeline scikit-learn (satu mesin) — pembanding"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3ff73c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = train_test_split(df, test_size=0.2, random_state=42, stratify=df.churn)\n",
    "pre = ColumnTransformer([(\"num\", StandardScaler(), FITUR),\n",
    "                         (\"kat\", OneHotEncoder(), [\"kontrak\"])])\n",
    "pipe = Pipeline([(\"pre\", pre), (\"lr\", LogisticRegression(max_iter=200))])\n",
    "t = time.time(); pipe.fit(train, train.churn)\n",
    "print(f\"Latih {time.time() - t:.3f}s | AUC uji: {roc_auc_score(test.churn, pipe.predict_proba(test)[:, 1]):.3f}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "65ef163b",
   "metadata": {},
   "source": [
    "## 6. Pipeline Spark MLlib\n",
    "\n",
    "Struktur sama: `StringIndexer → OneHotEncoder → VectorAssembler → StandardScaler → LogisticRegression`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "580d836f",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    from pyspark.ml import Pipeline as SparkPipeline\n",
    "    from pyspark.ml.feature import StringIndexer, OneHotEncoder as SOHE, VectorAssembler, StandardScaler as SScaler\n",
    "    from pyspark.ml.classification import LogisticRegression as SLR\n",
    "    from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "\n",
    "    sdf = spark.createDataFrame(df)\n",
    "    s_train, s_test = sdf.randomSplit([0.8, 0.2], seed=42)\n",
    "\n",
    "    stages = [\n",
    "        StringIndexer(inputCol=\"kontrak\", outputCol=\"kontrak_idx\"),\n",
    "        SOHE(inputCols=[\"kontrak_idx\"], outputCols=[\"kontrak_ohe\"]),\n",
    "        VectorAssembler(inputCols=FITUR + [\"kontrak_ohe\"], outputCol=\"fitur_mentah\"),\n",
    "        SScaler(inputCol=\"fitur_mentah\", outputCol=\"features\", withMean=True),\n",
    "        SLR(featuresCol=\"features\", labelCol=\"churn\", maxIter=50, regParam=0.01),\n",
    "    ]\n",
    "    model = SparkPipeline(stages=stages).fit(s_train)\n",
    "    pred = model.transform(s_test)\n",
    "    pred.select(\"churn\", \"probability\", \"prediction\").show(5, truncate=False)\n",
    "    auc = BinaryClassificationEvaluator(labelCol=\"churn\").evaluate(pred)\n",
    "    print(\"AUC Spark:\", round(auc, 3))\n",
    "    print(\"Koefisien:\", model.stages[-1].coefficients)\n",
    "else:\n",
    "    print(\"(dilewati — PySpark belum tersedia)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d36d6142",
   "metadata": {},
   "source": [
    "### Cross-validation di Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d3fd6d9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "    lr_stage = stages[-1]\n",
    "    grid = ParamGridBuilder().addGrid(lr_stage.regParam, [0.001, 0.01, 0.1]).build()\n",
    "    cv = CrossValidator(estimator=SparkPipeline(stages=stages), estimatorParamMaps=grid,\n",
    "                        evaluator=BinaryClassificationEvaluator(labelCol=\"churn\"),\n",
    "                        numFolds=3, parallelism=2)\n",
    "    cvm = cv.fit(s_train)\n",
    "    print(\"AUC rata-rata per regParam:\", [round(a, 4) for a in cvm.avgMetrics])\n",
    "else:\n",
    "    print(\"(dilewati — PySpark belum tersedia)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae196ee5",
   "metadata": {},
   "source": [
    "### K-Means MLlib pada data segmentasi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "943ccfdf",
   "metadata": {},
   "outputs": [],
   "source": [
    "seg = segmentasi_pelanggan()\n",
    "FS = [\"recency_hari\", \"frekuensi\", \"monetary_ribu\"]\n",
    "if ADA_SPARK:\n",
    "    from pyspark.ml.clustering import KMeans as SKMeans\n",
    "    from pyspark.ml.evaluation import ClusteringEvaluator\n",
    "    ssdf = spark.createDataFrame(seg)\n",
    "    km_pipe = SparkPipeline(stages=[\n",
    "        VectorAssembler(inputCols=FS, outputCol=\"raw\"),\n",
    "        SScaler(inputCol=\"raw\", outputCol=\"features\", withMean=True),\n",
    "        SKMeans(k=4, seed=1, featuresCol=\"features\"),\n",
    "    ]).fit(ssdf)\n",
    "    hasil = km_pipe.transform(ssdf)\n",
    "    print(\"Silhouette:\", round(ClusteringEvaluator().evaluate(hasil), 3))\n",
    "    hasil.groupBy(\"prediction\").agg(*[F.round(F.avg(c), 1).alias(c) for c in FS]).orderBy(\"prediction\").show()\n",
    "else:\n",
    "    Xs_seg = StandardScaler().fit_transform(seg[FS])\n",
    "    seg[\"klaster\"] = KMeans(4, n_init=10, random_state=1).fit_predict(Xs_seg)\n",
    "    print(\"(versi scikit-learn karena Spark belum tersedia)\")\n",
    "    print(seg.groupby(\"klaster\")[FS].mean().round(1))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7431ef0b",
   "metadata": {},
   "source": [
    "### FP-Growth di MLlib (menyambung Minggu 5)\n",
    "\n",
    "```python\n",
    "from pyspark.ml.fpm import FPGrowth\n",
    "from common.datasets import transaksi_ritel\n",
    "tdf = spark.createDataFrame([(i, t) for i, t in enumerate(transaksi_ritel())], [\"id\", \"items\"])\n",
    "fp = FPGrowth(itemsCol=\"items\", minSupport=0.05, minConfidence=0.4).fit(tdf)\n",
    "fp.freqItemsets.orderBy(F.desc(\"freq\")).show(10)\n",
    "fp.associationRules.orderBy(F.desc(\"lift\")).show(10)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "66e4af34",
   "metadata": {},
   "source": [
    "## ✍️ Coba sendiri\n",
    "\n",
    "1. Ubah jumlah partisi di bagian 3 (1, 3, 50). Apakah `w` berubah? Mengapa?\n",
    "2. Tambahkan regularisasi L2 ke gradient descent terdistribusi: `g += λ·w` (kecuali bias). Bagaimana efeknya ke koefisien?\n",
    "3. Implementasikan **mini-batch** / SGD terdistribusi: tiap iterasi hanya memakai sampel 10% data dari tiap partisi (seperti `miniBatchFraction`).\n",
    "4. Hitung **AUC** secara terdistribusi? Petunjuk: apakah AUC additive? Bandingkan dengan akurasi (jumlah benar per partisi)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "50b96159",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "311d22f3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 2."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7632921a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bad2eecb",
   "metadata": {},
   "outputs": [],
   "source": [
    "if ADA_SPARK:\n",
    "    spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "771cfd24",
   "metadata": {},
   "source": [
    "## Selanjutnya\n",
    "Kerjakan kuis dan `exercise.py`, lalu jalankan `python check.py 14` dari folder utama."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}