{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "ea83688f",
   "metadata": {},
   "source": [
    "# Lab Minggu 12 — HDFS & MapReduce (Simulasi Python)\n",
    "\n",
    "**Tujuan lab:**\n",
    "1. Menghitung kebutuhan blok & penyimpanan HDFS.\n",
    "2. Mensimulasikan job MapReduce: split → map → (combiner) → shuffle & sort → reduce.\n",
    "3. WordCount dan agregasi log transaksi.\n",
    "4. Melihat efek combiner & partitioner.\n",
    "5. (Opsional) Menulis mapper/reducer gaya **Hadoop Streaming**.\n",
    "\n",
    "Kita tidak butuh klaster Hadoop: semua dijalankan di Python agar alurnya terlihat."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f7e153b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.insert(0, \"..\")\n",
    "\n",
    "import math\n",
    "import zlib\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": "8571b05f",
   "metadata": {},
   "source": [
    "## 1. HDFS: blok & replikasi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f44728dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "def info_hdfs(ukuran_mb, blok_mb=128, replikasi=3):\n",
    "    blok = math.ceil(ukuran_mb / blok_mb)\n",
    "    return {\"ukuran_mb\": ukuran_mb, \"blok\": blok,\n",
    "            \"salinan_blok\": blok * replikasi, \"disk_mb\": ukuran_mb * replikasi,\n",
    "            \"blok_terakhir_mb\": ukuran_mb - (blok - 1) * blok_mb if blok else 0}\n",
    "\n",
    "pd.DataFrame([info_hdfs(x) for x in [100, 128, 500, 1024, 10_000]])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3528ada3",
   "metadata": {},
   "source": [
    "Simulasi penempatan replika (sederhana) di 6 DataNode dalam 2 rack:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c570b2a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "random.seed(0)\n",
    "rack = {\"dn1\": \"R1\", \"dn2\": \"R1\", \"dn3\": \"R1\", \"dn4\": \"R2\", \"dn5\": \"R2\", \"dn6\": \"R2\"}\n",
    "\n",
    "def tempatkan(blok_id, penulis=\"dn1\"):\n",
    "    r1 = penulis\n",
    "    lain_rack = [d for d in rack if rack[d] != rack[r1]]\n",
    "    r2 = random.choice(lain_rack)\n",
    "    r3 = random.choice([d for d in lain_rack if d != r2])\n",
    "    return {\"blok\": blok_id, \"replika\": [r1, r2, r3], \"rack\": [rack[r1], rack[r2], rack[r3]]}\n",
    "\n",
    "pd.DataFrame([tempatkan(f\"B{i}\") for i in range(4)])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ee96eafd",
   "metadata": {},
   "source": [
    "✍️ Jika `dn1` mati, apakah semua blok masih punya salinan? Jika seluruh rack R2 mati?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24d9363f",
   "metadata": {},
   "source": [
    "## 2. Mesin MapReduce mini"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aa97ff38",
   "metadata": {},
   "outputs": [],
   "source": [
    "def bagi_split(data, n_split):\n",
    "    q, r = divmod(len(data), n_split)\n",
    "    hasil, i = [], 0\n",
    "    for s in range(n_split):\n",
    "        uk = q + (1 if s < r else 0)\n",
    "        hasil.append(data[i:i + uk]); i += uk\n",
    "    return hasil\n",
    "\n",
    "def shuffle_sort(pasangan):\n",
    "    grup = {}\n",
    "    for k, v in pasangan:\n",
    "        grup.setdefault(k, []).append(v)\n",
    "    return sorted(grup.items())\n",
    "\n",
    "def mapreduce(data, mapper, reducer, combiner=None, n_split=4, verbose=False):\n",
    "    keluaran_map = []\n",
    "    for i, split in enumerate(bagi_split(data, n_split)):\n",
    "        out = [kv for rec in split for kv in mapper(rec)]\n",
    "        if combiner:\n",
    "            out = [combiner(k, vs) for k, vs in shuffle_sort(out)]\n",
    "        if verbose:\n",
    "            print(f\"mapper {i}: {len(split)} record -> {len(out)} pasangan  {out[:4]}...\")\n",
    "        keluaran_map.extend(out)\n",
    "    grup = shuffle_sort(keluaran_map)\n",
    "    if verbose:\n",
    "        print(\"setelah shuffle & sort:\", grup[:3], \"...\")\n",
    "    return dict(reducer(k, vs) for k, vs in grup)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d1f56a15",
   "metadata": {},
   "source": [
    "## 3. WordCount"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46c6f8b8",
   "metadata": {},
   "outputs": [],
   "source": [
    "docs = dokumen_teks()\n",
    "for d in docs[:3]:\n",
    "    print(\"-\", d)\n",
    "\n",
    "def mapper_wc(doc):\n",
    "    return [(w, 1) for w in doc.lower().split()]\n",
    "\n",
    "def reducer_sum(k, vs):\n",
    "    return (k, sum(vs))\n",
    "\n",
    "wc = mapreduce(docs, mapper_wc, reducer_sum, n_split=3, verbose=True)\n",
    "pd.Series(wc).sort_values(ascending=False).head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5713fdb1",
   "metadata": {},
   "outputs": [],
   "source": [
    "assert wc == dict(Counter(w for d in docs for w in d.split()))\n",
    "print(\"Hasil sama dengan Counter ✔\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "acad504b",
   "metadata": {},
   "source": [
    "## 4. Efek combiner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "039bb3b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "def hitung_pasangan(data, mapper, combiner=None, n_split=4):\n",
    "    total = 0\n",
    "    for split in bagi_split(data, n_split):\n",
    "        out = [kv for rec in split for kv in mapper(rec)]\n",
    "        if combiner:\n",
    "            out = [combiner(k, vs) for k, vs in shuffle_sort(out)]\n",
    "        total += len(out)\n",
    "    return total\n",
    "\n",
    "log = log_transaksi()\n",
    "mapper_kota = lambda b: [(b.split(\",\")[1], int(b.split(\",\")[3]) * int(b.split(\",\")[4]))]\n",
    "\n",
    "for n_split in [2, 8, 32]:\n",
    "    a = hitung_pasangan(log, mapper_kota, None, n_split)\n",
    "    b = hitung_pasangan(log, mapper_kota, reducer_sum, n_split)\n",
    "    print(f\"{n_split:>3} split: tanpa combiner {a:>5} pasangan | dengan combiner {b:>4} pasangan\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25c551a4",
   "metadata": {},
   "source": [
    "Dengan combiner, tiap mapper hanya mengirim satu pasangan per kota (6 kota) → hemat jaringan drastis.\n",
    "\n",
    "## 5. Agregasi log transaksi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b803c6dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(log[:3])\n",
    "pendapatan = mapreduce(log, mapper_kota, reducer_sum, combiner=reducer_sum)\n",
    "s = pd.Series(pendapatan).sort_values()\n",
    "s.plot.barh(title=\"Total pendapatan per kota (MapReduce)\")\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3f439b6",
   "metadata": {},
   "source": [
    "Bandingkan dengan pandas (`GROUP BY`):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f8a716b6",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.DataFrame([b.split(\",\") for b in log], columns=[\"tanggal\", \"kota\", \"produk\", \"jumlah\", \"harga\"])\n",
    "df[\"pendapatan\"] = df[\"jumlah\"].astype(int) * df[\"harga\"].astype(int)\n",
    "acuan = df.groupby(\"kota\")[\"pendapatan\"].sum()\n",
    "print((acuan == pd.Series(pendapatan)[acuan.index]).all())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "84eb7318",
   "metadata": {},
   "source": [
    "### Rata-rata: combiner yang salah vs benar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3833d144",
   "metadata": {},
   "outputs": [],
   "source": [
    "mapper_qty = lambda b: [(b.split(\",\")[2], int(b.split(\",\")[3]))]\n",
    "avg = lambda k, vs: (k, sum(vs) / len(vs))\n",
    "salah = mapreduce(log, mapper_qty, avg, combiner=avg, n_split=7)\n",
    "\n",
    "mapper_qty2 = lambda b: [(b.split(\",\")[2], (int(b.split(\",\")[3]), 1))]\n",
    "gabung = lambda k, vs: (k, (sum(v[0] for v in vs), sum(v[1] for v in vs)))\n",
    "benar = {k: t / n for k, (t, n) in mapreduce(log, mapper_qty2, gabung, combiner=gabung, n_split=7).items()}\n",
    "\n",
    "df[\"jumlah\"] = df[\"jumlah\"].astype(int)\n",
    "cmp = pd.DataFrame({\"pandas\": df.groupby(\"produk\")[\"jumlah\"].mean(),\n",
    "                    \"combiner_salah\": pd.Series(salah), \"combiner_benar\": pd.Series(benar)})\n",
    "cmp.head(6)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46003908",
   "metadata": {},
   "source": [
    "## 6. Partitioner: key mana ke reducer mana?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "846216d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "R = 3\n",
    "part = {k: zlib.crc32(k.encode()) % R for k in pendapatan}\n",
    "pd.Series(part, name=\"reducer\").sort_values()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e132702e",
   "metadata": {},
   "source": [
    "✍️ Apakah beban tiap reducer seimbang? Apa yang terjadi bila satu key sangat dominan (*data skew*)?\n",
    "\n",
    "## 7. (Opsional) Hadoop Streaming\n",
    "\n",
    "Hadoop Streaming menjalankan program apa pun yang membaca **stdin** dan menulis **stdout**\n",
    "(format `key\\tvalue`). Simpan dua file berikut lalu jalankan di klaster:\n",
    "\n",
    "```bash\n",
    "hadoop jar hadoop-streaming.jar -input /log -output /out \\\n",
    "    -mapper mapper.py -reducer reducer.py -file mapper.py -file reducer.py\n",
    "```\n",
    "\n",
    "Kita bisa mengujinya secara lokal dengan pipeline Unix: `cat log.csv | python mapper.py | sort | python reducer.py`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "244d436a",
   "metadata": {},
   "outputs": [],
   "source": [
    "MAPPER = '''import sys\n",
    "for baris in sys.stdin:\n",
    "    tgl, kota, produk, jumlah, harga = baris.strip().split(\",\")\n",
    "    print(f\"{kota}\\\\t{int(jumlah) * int(harga)}\")\n",
    "'''\n",
    "REDUCER = '''import sys\n",
    "kota_lama, total = None, 0\n",
    "for baris in sys.stdin:                      # input SUDAH terurut per key\n",
    "    kota, nilai = baris.strip().split(\"\\\\t\")\n",
    "    if kota != kota_lama and kota_lama is not None:\n",
    "        print(f\"{kota_lama}\\\\t{total}\")\n",
    "        total = 0\n",
    "    kota_lama = kota\n",
    "    total += int(nilai)\n",
    "if kota_lama is not None:\n",
    "    print(f\"{kota_lama}\\\\t{total}\")\n",
    "'''\n",
    "print(MAPPER)\n",
    "print(REDUCER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6383aa7d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Simulasi pipeline `mapper | sort | reducer` tanpa menulis file\n",
    "import io, contextlib\n",
    "\n",
    "def jalankan_skrip(kode, teks_input):\n",
    "    keluar = io.StringIO()\n",
    "    with contextlib.redirect_stdout(keluar):\n",
    "        lama = sys.stdin\n",
    "        sys.stdin = io.StringIO(teks_input)\n",
    "        try:\n",
    "            exec(kode, {})\n",
    "        finally:\n",
    "            sys.stdin = lama\n",
    "    return keluar.getvalue()\n",
    "\n",
    "keluaran_map = jalankan_skrip(MAPPER, \"\\n\".join(log))\n",
    "terurut = \"\".join(sorted(keluaran_map.splitlines(keepends=True)))\n",
    "print(jalankan_skrip(REDUCER, terurut))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9110ce19",
   "metadata": {},
   "source": [
    "## ✍️ Coba sendiri\n",
    "\n",
    "1. Hitung **jumlah transaksi per produk per bulan** (key = `(bulan, produk)`).\n",
    "2. Cari **produk terlaris tiap kota** (butuh dua tahap MapReduce atau reducer yang lebih pintar).\n",
    "3. Inverted index: untuk setiap kata di `dokumen_teks()`, daftar indeks dokumen yang memuatnya.\n",
    "4. Berapa blok & disk yang dibutuhkan untuk menyimpan 50 GB log per hari selama 1 tahun dengan replikasi 3?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "80788ed9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f875b61",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 2."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3563750f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "49344a0b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 4."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6c0e3e27",
   "metadata": {},
   "source": [
    "## Selanjutnya\n",
    "Kerjakan kuis dan `exercise.py`, lalu jalankan `python check.py 12` dari folder utama."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}