EQ_ykv_output.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://WIN10:4043\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>EQ_ykv</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x1adac008a58>"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "data size: \n",
      "\n",
      "sample :  ['_ID', 'TimeSt', 'Country', 'Province', 'City', 'Latitude', 'Longitude']\n",
      "(22025, 7)\n",
      "pois :  ['POIID', 'Latitude', 'Longitude']\n",
      "(4, 3) \n",
      "\n",
      "Duplicated data - time stamp and location : \n",
      "\n",
      "2026\n",
      "Number of countries: \n",
      "\n",
      "Cleaned records: \n",
      " 19871\n",
      "+-------+--------------------+---------+---------------+------------------+\n",
      "|    _ID|              TimeSt|     City|            POI|           minDist|\n",
      "+-------+--------------------+---------+---------------+------------------+\n",
      "|4523251|2017-06-21 19:00:...|   Guelph|POI2 - MONTREAL| 583.1789340044905|\n",
      "|4530411|2017-06-21 04:00:...|Thornhill|POI2 - MONTREAL| 502.5864355114627|\n",
      "|4544563|2017-06-21 15:01:...|  Toronto|POI2 - MONTREAL|501.54997848296216|\n",
      "|4550906|2017-06-21 05:01:...|   London|POI2 - MONTREAL| 683.3587874652333|\n",
      "|4573277|2017-06-21 09:01:...|  Renfrew|POI2 - MONTREAL| 243.2407245730464|\n",
      "|4575633|2017-06-21 13:02:...|Vancouver|POI1 - EDMONTON| 845.8776323682429|\n",
      "|4582656|2017-06-21 17:02:...|  Calgary|POI1 - EDMONTON|281.78694962282947|\n",
      "|4584527|2017-06-21 15:02:...|    Laval|POI2 - MONTREAL| 19.74598382920213|\n",
      "|4584920|2017-06-21 10:02:...| Wrentham|POI1 - EDMONTON|464.78356080913017|\n",
      "|4585618|2017-06-21 05:02:...|  Calgary|POI1 - EDMONTON|293.85650299885936|\n",
      "|4586259|2017-06-21 08:02:...|  Lacombe|POI1 - EDMONTON|121.32335536718419|\n",
      "|4602217|2017-06-21 09:04:...|  Toronto|POI2 - MONTREAL| 510.8201388633522|\n",
      "|4614987|2017-06-21 08:05:...|   Milton|POI2 - MONTREAL| 555.9820409522475|\n",
      "|4637235|2017-06-21 16:06:...|Newmarket|POI2 - MONTREAL|500.00799128225253|\n",
      "|4637946|2017-06-21 03:06:...|   Ottawa|POI2 - MONTREAL|171.12503931738408|\n",
      "|4663453|2017-06-21 16:08:...| Red Deer|POI1 - EDMONTON|146.80490191256672|\n",
      "|4664983|2017-06-21 16:08:...|   Oshawa|POI2 - MONTREAL|460.99006328050314|\n",
      "|4691479|2017-06-21 00:10:...|  Calgary|POI1 - EDMONTON|  285.196423291003|\n",
      "|4697224|2017-06-21 19:11:...|  Calgary|POI1 - EDMONTON| 279.9691099779517|\n",
      "|4703272|2017-06-21 15:11:...| Montréal|POI2 - MONTREAL|1.3899278655994876|\n",
      "+-------+--------------------+---------+---------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "+------------------+------------------+\n",
      "|               POI|      avg(minDist)|\n",
      "+------------------+------------------+\n",
      "|   POI2 - MONTREAL|458.34837448221276|\n",
      "|   POI1 - EDMONTON|294.85557983841613|\n",
      "|POI3 - NOVA SCOTIA|146.49472942297874|\n",
      "+------------------+------------------+\n",
      "\n",
      "+------------------+--------------------+\n",
      "|               POI|stddev_samp(minDist)|\n",
      "+------------------+--------------------+\n",
      "|   POI2 - MONTREAL|    228.108239364406|\n",
      "|   POI1 - EDMONTON|   287.6127766447783|\n",
      "|POI3 - NOVA SCOTIA|   79.89855056387565|\n",
      "+------------------+--------------------+\n",
      "\n",
      "+------------------+-------------------+\n",
      "|               POI|       min(minDist)|\n",
      "+------------------+-------------------+\n",
      "|   POI2 - MONTREAL| 0.8093687171976599|\n",
      "|   POI1 - EDMONTON|0.34844834633502614|\n",
      "|POI3 - NOVA SCOTIA| 16.227691135268355|\n",
      "+------------------+-------------------+\n",
      "\n",
      "+------------------+------------------+\n",
      "|               POI|      max(minDist)|\n",
      "+------------------+------------------+\n",
      "|   POI2 - MONTREAL|1497.6822580304956|\n",
      "|   POI1 - EDMONTON|1483.0054158570954|\n",
      "|POI3 - NOVA SCOTIA| 558.6544170261211|\n",
      "+------------------+------------------+\n",
      "\n",
      "+------------------+-------------------+\n",
      "|               POI|  skewness(minDist)|\n",
      "+------------------+-------------------+\n",
      "|   POI2 - MONTREAL|0.06694264070652206|\n",
      "|   POI1 - EDMONTON| 1.4241489936386473|\n",
      "|POI3 - NOVA SCOTIA| 0.9532915462390601|\n",
      "+------------------+-------------------+\n",
      "\n",
      "+------------------+------------------+\n",
      "|               POI| kurtosis(minDist)|\n",
      "+------------------+------------------+\n",
      "|   POI2 - MONTREAL| 2.331133468737165|\n",
      "|   POI1 - EDMONTON|2.0224187402527862|\n",
      "|POI3 - NOVA SCOTIA|1.9436436818666118|\n",
      "+------------------+------------------+\n",
      "\n",
      "+--------+------------------+------------------+------------------+------------------+-----------------------+\n",
      "|Requests|               POI|              Mean|            Median|      poiRadius_km|Density_Requests_by_km2|\n",
      "+--------+------------------+------------------+------------------+------------------+-----------------------+\n",
      "|    9816|   POI2 - MONTREAL|458.34837448221276|511.67044847139715|1497.6822580304956|   0.001392982542784...|\n",
      "|    9695|   POI1 - EDMONTON|294.85557983841613|277.56092637638926|1483.0054158570954|   0.001403178215004784|\n",
      "|     360|POI3 - NOVA SCOTIA|146.49472942297874| 152.5009259557654| 558.6544170261211|   3.671694349361896E-4|\n",
      "+--------+------------------+------------------+------------------+------------------+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from __future__ import print_function\n",
    "import numpy as np\n",
    "import pyspark as ps\n",
    "import math\n",
    "import os\n",
    "import urllib\n",
    "import sys\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = ps.sql.SparkSession.builder.appName('EQ_ykv').getOrCreate()\n",
    "\n",
    "# Load data \n",
    "\n",
    "sample = spark.read\\\n",
    "  .format('org.apache.spark.sql.execution.datasources.csv.CSVFileFormat')\\\n",
    "  .option('header', 'true')\\\n",
    "  .load('data/DataSample.csv')\n",
    "\n",
    "\n",
    "poi = spark.read\\\n",
    "  .format('org.apache.spark.sql.execution.datasources.csv.CSVFileFormat')\\\n",
    "  .option('header', 'true')\\\n",
    "  .load('data/POIList.csv')\n",
    "\n",
    "\n",
    "# Stage 0 - CLEANUP\n",
    "\n",
    "# Removing spaces from names\n",
    "sample = sample.withColumnRenamed(\" TimeSt\", \"TimeSt\")\n",
    "poi = poi.withColumnRenamed(\" Latitude\", \"Latitude\")\n",
    "\n",
    "# data size\n",
    "\n",
    "print(\"data size: \\n\")\n",
    "print(\"sample : \", sample.columns)\n",
    "print((sample.count(), len(sample.columns)))\n",
    "print(\"pois : \", poi.columns)\n",
    "print((poi.count(), len(poi.columns)), \"\\n\")\n",
    "\n",
    "# Temp tables\n",
    "sample.registerTempTable(\"sample0\")\n",
    "poi.registerTempTable(\"pois0\")\n",
    "\n",
    "# duplicate removal \n",
    "\n",
    "query = \"\"\"SELECT *\n",
    "FROM sample0 A\n",
    "WHERE _ID > (SELECT MIN(_ID) FROM sample0 B\n",
    "WHERE A.TimeSt = B.TimeSt AND A.Latitude=B.Latitude AND A.Longitude=B.Longitude)\"\"\"\n",
    "\n",
    "duplicates = spark.sql(query)\n",
    "print(\"Duplicated data - time stamp and location : \\n\")\n",
    "print(duplicates.count())\n",
    "\n",
    "clean = sample.subtract(duplicates)\n",
    "\n",
    "# Temp table\n",
    "clean.registerTempTable(\"clean0\")\n",
    "\n",
    "# Canada Only data\n",
    "print(\"Number of countries: \\n\")\n",
    "clean.select(\"Country\").distinct().count()\n",
    "\n",
    "# Removing mislabeled data \n",
    "\n",
    "query = \"\"\" SELECT *\n",
    "FROM clean0\n",
    "WHERE Latitude > 40 AND Longitude > -130 AND Longitude < -60\n",
    "\"\"\"\n",
    "\n",
    "clean2 = spark.sql(query)\n",
    "print(\"Cleaned records: \\n\",clean2.count())\n",
    "# Temp table\n",
    "clean2.registerTempTable(\"clean2\")\n",
    "\n",
    "# 2 out of four POIs where the same / Location convertion to radians\n",
    "\n",
    "poi = poi.withColumn(\"lat_rad\", poi.Latitude* math.pi / 180)\n",
    "poi = poi.withColumn(\"lon_rad\", poi.Longitude* math.pi / 180)\n",
    "\n",
    "\n",
    "# Saage 1 - minimum distance and POI Assignation\n",
    "\n",
    "# Distance calculation and location conversion to radians\n",
    "\n",
    "lat0 = F.toRadians(\"Latitude\").alias(\"lat0\")\n",
    "lon0 = F.toRadians(\"Longitude\").alias(\"lon0\")\n",
    "lat1 = 0.9345569159727344\n",
    "lon1 = -1.9806997123424743\n",
    "lat2 = 0.7945023069213337\n",
    "lon2 = -1.2839693364011688\n",
    "lat3 = 0.7893221871547071\n",
    "lon3 = -1.1036193160713015\n",
    "dlat1 = lat1 - lat0\n",
    "dlon1 = lon1 - lon0\n",
    "dlat2 = lat2 - lat0\n",
    "dlon2 = lon2 - lon0\n",
    "dlat3 = lat3 - lat0\n",
    "dlon3 = lon3 - lon0\n",
    "a1 = F.sin(dlat1/2)**2 + F.cos(lat0) * F.cos(lat0) * F.sin(dlon1/2)**2\n",
    "a2 = F.sin(dlat2/2)**2 + F.cos(lat0) * F.cos(lat0) * F.sin(dlon2/2)**2\n",
    "a3 = F.sin(dlat3/2)**2 + F.cos(lat0) * F.cos(lat0) * F.sin(dlon3/2)**2\n",
    "c1 = F.lit(2) * F.asin(F.sqrt(a1)) \n",
    "c2 = F.lit(2) * F.asin(F.sqrt(a2)) \n",
    "c3 = F.lit(2) * F.asin(F.sqrt(a3)) \n",
    "r = F.lit(6371)\n",
    "dist1 = (c1 * r).alias('dist1')\n",
    "dist2 = (c2 * r).alias('dist2')\n",
    "dist3 = (c3 * r).alias('dist3')\n",
    "\n",
    "distances = clean2.select(\"_ID\", \"TimeSt\", \"City\", \"Province\",\"Latitude\", \"Longitude\", dist1, dist2, dist3)\n",
    "distances.registerTempTable(\"dist0\")\n",
    "\n",
    "# POI assignation and minimal distance to poi\n",
    "\n",
    "query = \"\"\"SELECT _ID,  TimeSt, City, Province, dist1, dist2, dist3,\n",
    "    CASE WHEN (dist1 < dist2) AND (dist1 < dist3) THEN \"POI1 - EDMONTON\" \n",
    "         WHEN (dist2 < dist1) AND (dist2 < dist3) THEN \"POI2 - MONTREAL\" \n",
    "         ELSE \"POI3 - NOVA SCOTIA\" \n",
    "         END AS POI\n",
    "FROM dist0\n",
    "\n",
    "\"\"\"\n",
    "\n",
    "distPOI = spark.sql(query)\n",
    "distPOI.registerTempTable(\"distPOI0\")\n",
    "\n",
    "query = \"\"\"SELECT _ID, TimeSt, City,  POI,\n",
    "    CASE WHEN (dist1 < dist2) AND (dist1 < dist3) THEN dist1 \n",
    "         WHEN (dist2 < dist1) AND (dist2 < dist3) THEN dist2\n",
    "         ELSE dist3 \n",
    "         END AS minDist\n",
    "FROM distPOI0\n",
    "\"\"\"\n",
    "distPOI2 = spark.sql(query)\n",
    "distPOI2.registerTempTable(\"distPOI2\")\n",
    "distPOI2.show()\n",
    "\n",
    "\n",
    "# Stage 2 Analysis\n",
    "\n",
    "# grouping data by POI\n",
    "by_POI = distPOI2.groupBy(\"POI\")\n",
    "\n",
    "\n",
    "by_POI.avg(\"minDist\").show()\n",
    "\n",
    "by_POI.agg(F.stddev(\"minDist\")).show()\n",
    "\n",
    "by_POI.min(\"minDist\").show()\n",
    "\n",
    "by_POI.max(\"minDist\").show()\n",
    "\n",
    "by_POI.agg(F.skewness(\"minDist\")).show()\n",
    "\n",
    "by_POI.agg(F.kurtosis(\"minDist\")).show()\n",
    "\n",
    "\n",
    "query = \"\"\"SELECT COUNT(_ID) Requests, POI, AVG(minDist) AS Mean,  percentile_approx(minDist, 0.5) AS Median,\n",
    "MAX(minDist) AS poiRadius_km, COUNT(_ID)/(3.14159*POWER(MAX(minDist),2)) AS Density_Requests_by_km2\n",
    "FROM distPOI2\n",
    "GROUP BY POI\n",
    "\"\"\"\n",
    "spark.sql(query).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}