validation_another.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 97,
   "id": "fd79a247",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import StructType, DoubleType, StringType, IntegerType\n",
    "import math\n",
    "from operator import add\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "id": "e76357b6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 77:===========================================>              (3 + 1) / 4]\r",
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "1587257"
      ]
     },
     "execution_count": 125,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark = SparkSession.builder.appName(\"KMeansWithMapReduce\").getOrCreate()\n",
    "\n",
    "schema = StructType() \\\n",
    "    .add(\"rowID\", IntegerType(), True) \\\n",
    "    .add(\"hpwren_timestamp\", StringType(), True) \\\n",
    "    .add(\"air_pressure\", DoubleType(), True) \\\n",
    "    .add(\"air_temp\", DoubleType(), True) \\\n",
    "    .add(\"avg_wind_direction\", DoubleType(), True) \\\n",
    "    .add(\"avg_wind_speed\", DoubleType(), True) \\\n",
    "    .add(\"max_wind_direction\", DoubleType(), True) \\\n",
    "    .add(\"max_wind_speed\", DoubleType(), True) \\\n",
    "    .add(\"min_wind_direction\", DoubleType(), True) \\\n",
    "    .add(\"min_wind_speed\", DoubleType(), True) \\\n",
    "    .add(\"rain_accumulation\", DoubleType(), True) \\\n",
    "    .add(\"rain_duration\", DoubleType(), True) \\\n",
    "    .add(\"relative_humidity\", DoubleType(), True)\n",
    "\n",
    "df = spark.read.format(\"csv\").option(\"header\", True).schema(schema).load(\"./data/minute_weather.csv\")\n",
    "df.na.drop()\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "id": "a4708367",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------------------+------------+--------+------------------+--------------+------------------+--------------+------------------+--------------+-----------------+-------------+-----------------+\n",
      "|rowID|   hpwren_timestamp|air_pressure|air_temp|avg_wind_direction|avg_wind_speed|max_wind_direction|max_wind_speed|min_wind_direction|min_wind_speed|rain_accumulation|rain_duration|relative_humidity|\n",
      "+-----+-------------------+------------+--------+------------------+--------------+------------------+--------------+------------------+--------------+-----------------+-------------+-----------------+\n",
      "|    0|2011-09-10 00:00:49|       912.3|   64.76|              97.0|           1.2|             106.0|           1.6|              85.0|           1.0|             null|         null|             60.5|\n",
      "|    1|2011-09-10 00:01:49|       912.3|   63.86|             161.0|           0.8|             215.0|           1.5|              43.0|           0.2|              0.0|          0.0|             39.9|\n",
      "|    2|2011-09-10 00:02:49|       912.3|   64.22|              77.0|           0.7|             143.0|           1.2|             324.0|           0.3|              0.0|          0.0|             43.0|\n",
      "|    3|2011-09-10 00:03:49|       912.3|    64.4|              89.0|           1.2|             112.0|           1.6|              12.0|           0.7|              0.0|          0.0|             49.5|\n",
      "|    4|2011-09-10 00:04:49|       912.3|    64.4|             185.0|           0.4|             260.0|           1.0|             100.0|           0.1|              0.0|          0.0|             58.8|\n",
      "|    5|2011-09-10 00:05:49|       912.3|    63.5|              76.0|           2.5|              92.0|           3.0|              61.0|           2.0|              0.0|          0.0|             62.6|\n",
      "|    6|2011-09-10 00:06:49|       912.3|   62.78|              79.0|           2.4|              89.0|           2.7|              62.0|           2.0|              0.0|          0.0|             65.6|\n",
      "|    7|2011-09-10 00:07:49|       912.3|   62.42|              86.0|           2.0|              92.0|           2.4|              75.0|           1.8|              0.0|          0.0|             65.2|\n",
      "|    8|2011-09-10 00:08:49|       912.3|   62.24|             105.0|           1.4|             125.0|           1.9|              82.0|           1.0|              0.0|          0.0|             65.8|\n",
      "|    9|2011-09-10 00:09:49|       912.3|   62.24|              93.0|           0.4|             126.0|           0.7|              14.0|           0.2|              0.0|          0.0|             58.6|\n",
      "|   10|2011-09-10 00:10:49|       912.3|   62.24|             144.0|           1.2|             167.0|           1.8|             115.0|           0.6|              0.0|          0.0|             38.5|\n",
      "|   11|2011-09-10 00:11:49|       912.2|   63.14|             105.0|           1.6|             126.0|           2.0|              92.0|           0.9|              0.0|          0.0|             42.6|\n",
      "|   12|2011-09-10 00:12:49|       912.2|   64.04|             116.0|           1.8|             143.0|           2.7|             104.0|           1.1|              0.0|          0.0|             45.3|\n",
      "|   13|2011-09-10 00:13:49|       912.2|    64.4|             142.0|           1.1|             200.0|           1.9|              93.0|           0.7|              0.0|          0.0|             36.1|\n",
      "|   14|2011-09-10 00:14:49|       912.2|   64.94|             150.0|           1.3|             173.0|           2.1|             117.0|           0.8|              0.0|          0.0|             33.2|\n",
      "|   15|2011-09-10 00:15:49|       912.2|   65.48|              90.0|           1.5|             100.0|           1.9|              80.0|           1.3|              0.0|          0.0|             45.2|\n",
      "|   16|2011-09-10 00:16:49|       912.3|   65.48|              91.0|           0.7|             168.0|           1.2|              26.0|           0.4|              0.0|          0.0|             55.9|\n",
      "|   17|2011-09-10 00:17:49|       912.3|   64.94|              77.0|           1.7|              91.0|           2.2|              55.0|           0.8|              0.0|          0.0|             60.5|\n",
      "|   18|2011-09-10 00:18:49|       912.3|   64.04|             107.0|           1.9|             116.0|           2.4|              95.0|           1.5|              0.0|          0.0|             61.8|\n",
      "|   19|2011-09-10 00:19:49|       912.2|   63.32|             109.0|           0.9|             119.0|           1.3|              98.0|           0.6|              0.0|          0.0|             56.0|\n",
      "+-----+-------------------+------------+--------+------------------+--------------+------------------+--------------+------------------+--------------+-----------------+-------------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "id": "c5d6b297",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# k clusters\n",
    "k = 3\n",
    "# threshold between new and old centroid\n",
    "threshold = 0.4\n",
    "\n",
    "# randomly select centroids\n",
    "centroids = df.rdd.takeSample(False, k, seed=0)\n",
    "centroids = [[c.air_pressure, c.air_temp, c.relative_humidity] for c in centroids]\n",
    "# centroid structure: (index, [centroid features]) e.g., (1, [2.0, 3.0, 5.0, 0.2])\n",
    "centroids = [(idx, centroid) for idx, centroid in enumerate(centroids)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "id": "b7fc6767",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0, [920.5, 39.02, 89.2])\n",
      "(1, [913.8, 68.18, 24.9])\n",
      "(2, [915.9, 69.98, 38.6])\n"
     ]
    }
   ],
   "source": [
    "# show centroid structure\n",
    "for _ in range(3):\n",
    "    print(centroids[_])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "id": "b90cf996",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[148] at RDD at PythonRDD.scala:53"
      ]
     },
     "execution_count": 102,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# set points\n",
    "points = df.rdd\n",
    "# point structure: ([point features], count) e.g., ([2.0, 5.0, 2.5, 0.6], 1)\n",
    "points_rdd = points.map(lambda p: ([p.air_pressure, p.air_temp, p.relative_humidity], 1))\n",
    "points_rdd.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "id": "a26fdf22",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[([912.3, 64.76, 60.5], 1),\n",
       " ([912.3, 63.86, 39.9], 1),\n",
       " ([912.3, 64.22, 43.0], 1)]"
      ]
     },
     "execution_count": 103,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "points_rdd.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "id": "e6b756f4",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\" calculate distance \"\"\"\n",
    "def calculateDistance(point, centroid):\n",
    "    distance = 0\n",
    "    for index in range(len(point)):\n",
    "        distance += (point[index]-centroid[index])**2\n",
    "    return math.sqrt(distance)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "id": "cb54c7a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\" belongs to Centroid \"\"\"\n",
    "def belongCluster(point, centroids):\n",
    "    centroidIndex = 0\n",
    "    closest = float(\"+inf\")\n",
    "    for centroid in centroids:\n",
    "        dist = calculateDistance(point, centroid[1])\n",
    "        if dist < closest:\n",
    "            closest = dist\n",
    "            centroidIndex = centroid[0]\n",
    "    return centroidIndex"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "id": "8b1c0540",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\" Reduce all points in each centroid \"\"\"\n",
    "def accumulatedCluster(p1, p2):\n",
    "    cluster_sum = list(map(add, p1[0], p2[0]))\n",
    "    cluster_count = p1[1]+p2[1]\n",
    "    p = (cluster_sum, cluster_count)\n",
    "    return p"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "id": "1020cb3c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Map Phase\n",
    "pointMapCentroid_rdd = points_rdd.keyBy(lambda point: belongCluster(point[0], centroids))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "id": "1b22e0a2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(2, ([912.3, 64.76, 60.5], 1)),\n",
       " (2, ([912.3, 63.86, 39.9], 1)),\n",
       " (2, ([912.3, 64.22, 43.0], 1)),\n",
       " (2, ([912.3, 64.4, 49.5], 1)),\n",
       " (2, ([912.3, 64.4, 58.8], 1))]"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pointMapCentroid_rdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "id": "0ff72b33",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reduce Phase\n",
    "pointReducedCentroid_rdd = pointMapCentroid_rdd.reduceByKey(lambda p1, p2: accumulatedCluster(p1, p2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "id": "549f34cb",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(0, ([387607269.6000054, 20669036.480000146, 35156074.10000006], 423277)),\n",
       " (1, ([511055543.3000114, 37387668.000000685, 10949992.000000026], 556539)),\n",
       " (2, ([556582223.49993, 40117427.29998607, 29460650.499999877], 607441))]"
      ]
     },
     "execution_count": 110,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pointReducedCentroid_rdd.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "id": "3e13c5a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "pointReducedCentroid_rdd = pointReducedCentroid_rdd.map(lambda p: (p[0], np.divide(p[1][0], p[1][1]).tolist()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "id": "377f1c10",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0, [915.7295803929942, 48.830993604661124, 83.05689678390289]),\n",
       " (1, [918.2744485112659, 67.17888234247857, 19.675156637719954]),\n",
       " (2, [916.2737179412158, 66.04333145109743, 48.49960819240037])]"
      ]
     },
     "execution_count": 112,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pointReducedCentroid_rdd.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "id": "f447e16f",
   "metadata": {},
   "outputs": [],
   "source": [
    "reduced_points = pointReducedCentroid_rdd.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "id": "c80dd311",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0, [915.7295803929942, 48.830993604661124, 83.05689678390289])\n",
      "(1, [918.2744485112659, 67.17888234247857, 19.675156637719954])\n",
      "(2, [916.2737179412158, 66.04333145109743, 48.49960819240037])\n"
     ]
    }
   ],
   "source": [
    "for _ in reduced_points:\n",
    "    print(_)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "id": "1a9f84ef",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new centroids\n",
    "new_centroids = sorted(reduced_points)\n",
    "centroids.sort()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "id": "32814be8",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0, [915.7295803929942, 48.830993604661124, 83.05689678390289]),\n",
       " (1, [918.2744485112659, 67.17888234247857, 19.675156637719954]),\n",
       " (2, [916.2737179412158, 66.04333145109743, 48.49960819240037])]"
      ]
     },
     "execution_count": 116,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# create new centroids\n",
    "new_centroids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "id": "288b7f6b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0, [920.5, 39.02, 89.2]),\n",
       " (1, [913.8, 68.18, 24.9]),\n",
       " (2, [915.9, 69.98, 38.6])]"
      ]
     },
     "execution_count": 117,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# create new centroids\n",
    "centroids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "id": "5cf047fc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "12.51999264621308\n",
      "6.951396565045956\n",
      "10.660171983910288\n",
      "not converged, still run\n"
     ]
    }
   ],
   "source": [
    "# check convergence or not\n",
    "convergence_percentage = 0\n",
    "for index, centroid in enumerate(centroids):\n",
    "    dist = calculateDistance(centroid[1], new_centroids[index][1])\n",
    "    print(dist)\n",
    "    \n",
    "    if dist < threshold:\n",
    "        convergence_percentage += 1\n",
    "        \n",
    "centroids = new_centroids\n",
    "percentage = len(centroids)*80/100\n",
    "\n",
    "if convergence_percentage > percentage:\n",
    "    print(\"Centroids converged\")\n",
    "else:\n",
    "    print(\"not converged, still run\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "id": "e636a643",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+------------+--------+-----------------+\n",
      "|cluster_index|air_pressure|air_temp|relative_humidity|\n",
      "+-------------+------------+--------+-----------------+\n",
      "|            2|       912.3|   64.76|             60.5|\n",
      "|            2|       912.3|   63.86|             39.9|\n",
      "|            2|       912.3|   64.22|             43.0|\n",
      "|            2|       912.3|    64.4|             49.5|\n",
      "|            2|       912.3|    64.4|             58.8|\n",
      "|            2|       912.3|    63.5|             62.6|\n",
      "|            2|       912.3|   62.78|             65.6|\n",
      "|            2|       912.3|   62.42|             65.2|\n",
      "|            2|       912.3|   62.24|             65.8|\n",
      "|            2|       912.3|   62.24|             58.6|\n",
      "|            2|       912.3|   62.24|             38.5|\n",
      "|            2|       912.2|   63.14|             42.6|\n",
      "|            2|       912.2|   64.04|             45.3|\n",
      "|            2|       912.2|    64.4|             36.1|\n",
      "|            1|       912.2|   64.94|             33.2|\n",
      "|            2|       912.2|   65.48|             45.2|\n",
      "|            2|       912.3|   65.48|             55.9|\n",
      "|            2|       912.3|   64.94|             60.5|\n",
      "|            2|       912.3|   64.04|             61.8|\n",
      "|            2|       912.2|   63.32|             56.0|\n",
      "+-------------+------------+--------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "schema_result = StructType() \\\n",
    "    .add(\"cluster_index\", IntegerType(), True) \\\n",
    "    .add(\"air_pressure\", DoubleType(), True) \\\n",
    "    .add(\"air_temp\", DoubleType(), True) \\\n",
    "    .add(\"relative_humidity\", DoubleType(), True)\n",
    "\n",
    "df_result1 = spark.createDataFrame(pointMapCentroid_rdd.map(lambda p: (p[0], p[1][0][0], p[1][0][1],\n",
    "                                                                    p[1][0][2])), schema=schema_result)\n",
    "df_result1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "id": "c58cd078",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_result1.write.option(\"header\", True).csv(\"./result_weather_1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "id": "a388517f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-----------------+------------------+------------------+\n",
      "|cluster_index|     air_pressure|          air_temp| relative_humidity|\n",
      "+-------------+-----------------+------------------+------------------+\n",
      "|            0|915.7295803929942|48.830993604661124| 83.05689678390289|\n",
      "|            1|918.2744485112659| 67.17888234247857|19.675156637719954|\n",
      "|            2|916.2737179412158| 66.04333145109743| 48.49960819240037|\n",
      "+-------------+-----------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_result2 = spark.createDataFrame(spark.sparkContext.parallelize(centroids).map(lambda p: (p[0], p[1][0], p[1][1],\n",
    "                                                    p[1][2])), schema=schema_result)\n",
    "df_result2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "id": "cce5cc3a",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_result2.coalesce(1).write.option(\"header\", True).csv(\"./result_weather_2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98d94a5b",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}