Distance_Analysis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DISTANCE ANALYSIS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook analyzes geographic data in BFKOORD_GEO.\n",
    "\n",
    "Following data structures created and saved as pickle file:\n",
    "- zurich_stations_set : set of stations around in Zurich (Circle of 10 km centered in Zurich HB)\n",
    "- distance_dictionary : dictionary that show the distance between two stations in Zurich\n",
    "- coordinate_dictionary : dictionary that gives latitude and longitude information of an arbitrary station in Zurich"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pickle\n",
    "import socket\n",
    "import getpass\n",
    "import os\n",
    "\n",
    "import ast\n",
    "import numpy as np\n",
    "from scipy import stats\n",
    "import pandas as pd\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "from math import sin, cos, sqrt, atan2, radians\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Change Layout to be able to see spark dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style>.container { width:95% !important; }</style>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "plt.rcParams['figure.figsize'] = (10,6)\n",
    "plt.rcParams['font.size'] = 18\n",
    "plt.style.use('fivethirtyeight')\n",
    "\n",
    "def fix_layout(width:int=95):\n",
    "    from IPython.core.display import display, HTML\n",
    "    display(HTML('<style>.container { width:' + str(width) + '% !important; }</style>'))\n",
    "    \n",
    "fix_layout()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "username = getpass.getuser()\n",
    "\n",
    "SPARK_LOCAL = False\n",
    "\n",
    "# on the laptop\n",
    "if not 'iccluster' in socket.gethostname():\n",
    "    # set this to the base spark directory on your system\n",
    "    SPARK_LOCAL = True\n",
    "    \n",
    "    if username == \"fatine\":\n",
    "        spark_home = '/home/fatine/spark-2.4.1-bin-hadoop2.7'\n",
    "        \n",
    "        try:\n",
    "            import findspark\n",
    "            findspark.init(spark_home)\n",
    "        except ModuleNotFoundError as e:\n",
    "            print('Info: {}'.format(e))\n",
    "    elif username == \"soner\":\n",
    "        spark_home = '/home/soner/Desktop/DSLAB2019/spark-2.4.1-bin-hadoop2.7'\n",
    "        \n",
    "        try:\n",
    "            import findspark\n",
    "            findspark.init(spark_home)\n",
    "        except ModuleNotFoundError as e:\n",
    "            print('Info: {}'.format(e))\n",
    "            \n",
    "    elif username == \"jelena\":\n",
    "        pass\n",
    "        \n",
    "    \n",
    "        \n",
    "# cluster\n",
    "if username == \"jbanjac\":\n",
    "    ROOT_PATH = \"/home/jbanjac/robust-journey-planning\"\n",
    "    os.environ['PYSPARK_PYTHON'] = '/opt/anaconda3/bin/python'\n",
    "# local\n",
    "elif username == \"jelena\":\n",
    "    ROOT_PATH = os.getcwd()\n",
    "    os.environ['PYSPARK_PYTHON'] = '/home/jelena/anaconda3/bin/python'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import unix_timestamp, udf, desc\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import Window\n",
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "from pyspark.ml.feature import VectorAssembler,Normalizer,  PCA,VectorIndexer\n",
    "from pyspark.ml.clustering import KMeans, BisectingKMeans, GaussianMixture"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "if username == \"jelena\":\n",
    "    spark = (SparkSession \\\n",
    "                .builder \\\n",
    "                .appName('sbb-{0}'.format(getpass.getuser())) \\\n",
    "                .master('local[4]') \\\n",
    "                .config('spark.driver.memory', '10g') \\\n",
    "                .config('spark.executor.memory', '4g') \\\n",
    "                .config('spark.executor.instances', '5') \\\n",
    "                .config('spark.port.maxRetries', '100') \\\n",
    "                .getOrCreate())\n",
    "    CLUSTER_URL = \"hdfs://iccluster042.iccluster.epfl.ch:8020\"\n",
    "\n",
    "elif SPARK_LOCAL:\n",
    "    spark = SparkSession \\\n",
    "                .builder \\\n",
    "                .master(\"local\") \\\n",
    "                .appName(\"roboust-journey-planing\") \\\n",
    "                .config(\"spark.driver.host\", \"localhost\") \\\n",
    "                .getOrCreate()\n",
    "    CLUSTER_URL = \"\"\n",
    "else:\n",
    "    spark = SparkSession \\\n",
    "                .builder \\\n",
    "                .master(\"yarn\") \\\n",
    "                .appName('sbb-{0}'.format(getpass.getuser())) \\\n",
    "                .config('spark.executor.memory', '4g') \\\n",
    "                .config('spark.executor.instances', '5') \\\n",
    "                .config('spark.port.maxRetries', '100') \\\n",
    "                .getOrCreate()\n",
    "    CLUSTER_URL = \"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "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://tsf-484-wpa-0-089.epfl.ch:4041\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[4]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>sbb-jelena</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f8db41213c8>"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc = spark.sparkContext\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- stop_number: string (nullable = true)\n",
      " |-- longitude: double (nullable = true)\n",
      " |-- latitude: double (nullable = true)\n",
      " |-- elevation: double (nullable = true)\n",
      " |-- stop_name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def read_hrdf(file_name):\n",
    "    with open(file_name, encoding='utf-8') as f:\n",
    "        lines = f.readlines()\n",
    "    \n",
    "    data = []\n",
    "    for line in lines:\n",
    "        data.append([line[:7].strip(), float(line[8:18].strip()), float(line[19:29].strip()), float(line[30:36].strip()), line[38:].strip()])\n",
    "    return data\n",
    "\n",
    "df = pd.DataFrame(data=read_hrdf('BFKOORD_GEO'), \n",
    "                       columns=[\"stop_number\", \"longitude\", \"latitude\", \"elevation\", \"stop_name\"])\n",
    "# convert to Spark DF\n",
    "df_geo = spark.createDataFrame(df)\n",
    "df_geo.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Stations Around Zurich"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select stations around 10 km of Zurich HB"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------+---------+---------+---------+\n",
      "|stop_number|longitude| latitude|elevation|stop_name|\n",
      "+-----------+---------+---------+---------+---------+\n",
      "|    8503000| 8.540192|47.378177|    408.0|Zürich HB|\n",
      "+-----------+---------+---------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_geo.where(df_geo['stop_name'].like('Zürich HB')).show()\n",
    "\n",
    "Zurich_HB_lat = df_geo.where(df_geo['stop_name'].like('Zürich HB')).\\\n",
    "                        select('latitude').collect()[0][0]\n",
    "Zurich_HB_lon = df_geo.where(df_geo['stop_name'].like('Zürich HB')).\\\n",
    "                        select('longitude').collect()[0][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "@F.udf\n",
    "def calculateDistance(latitude, longitude):\n",
    "    '''\n",
    "    The distance between a coordinate and Zurich HB, in kilometers.\n",
    "    '''\n",
    "    # approximate radius of earth in km\n",
    "    R = 6373.0\n",
    "\n",
    "    # latitude and longitude values of Zurich HB in terms of radians\n",
    "    lat_Zurich_HB = radians(Zurich_HB_lat)#radians(47.378178)\n",
    "    lon_Zurich_HB = radians(Zurich_HB_lon)#radians(8.540192)\n",
    "\n",
    "    # latitude and longitude values of a given station in terms of radians \n",
    "    # for comparing with the Zurich HB's coordinates\n",
    "    lon = radians(longitude)\n",
    "    lat = radians(latitude)\n",
    "    \n",
    "    # calculates the distance by using Haversine formula\n",
    "    dlon = lon - lon_Zurich_HB\n",
    "    dlat = lat - lat_Zurich_HB\n",
    "    a = sin(dlat / 2)**2 + cos(lat_Zurich_HB) * cos(lat) * sin(dlon / 2)**2\n",
    "    c = 2 * atan2(sqrt(a), sqrt(1 - a))\n",
    "\n",
    "    distance = R * c # as km\n",
    "    \n",
    "    return distance\n",
    "\n",
    "# Select the stations around 10 km of Zürich HB\n",
    "df_zurich_geo = df_geo.withColumn('distance_to_Zurich', calculateDistance(df_geo.latitude,df_geo.longitude)).filter('distance_to_Zurich<=10.0')#.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "create zurich_stations_set as set of strings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of stations around Zurich in BFKOORD_GEO is 1040\n"
     ]
    }
   ],
   "source": [
    "zurich_stations_set = set([row[0] for row in df_zurich_geo.select('stop_number').collect()])\n",
    "print('Number of stations around Zurich in BFKOORD_GEO is', len(zurich_stations_set))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Distances Between Stations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "@F.udf('float')\n",
    "def calculateCrossDistance(lat1, lon1, lat2, lon2):\n",
    "    '''\n",
    "    The distance between two stations, in meters.\n",
    "    '''\n",
    "    # approximate radius of earth in km\n",
    "    R = 6373.0\n",
    "\n",
    "    # latitude and longitude values of a given station in terms of radians \n",
    "    # for comparing with the Zurich HB's coordinates\n",
    "    lon_1 = radians(lon1)\n",
    "    lat_1 = radians(lat1)\n",
    "    lon_2 = radians(lon2)\n",
    "    lat_2 = radians(lat2)\n",
    "    \n",
    "    # calculates the distance by using Haversine formula\n",
    "    dlon = lon_2 - lon_1\n",
    "    dlat = lat_2 - lat_1\n",
    "    a = sin(dlat / 2)**2 + cos(lat_1) * cos(lat_2) * sin(dlon / 2)**2\n",
    "    c = 2 * atan2(sqrt(a), sqrt(1 - a))\n",
    "\n",
    "    distance = R * c * 1000 # as m\n",
    "    \n",
    "    return distance"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "calculate the distance between any two stations in Zurich"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "walk_from_df = df_zurich_geo.alias('walk_from_df').withColumnRenamed('longitude', 'longitude_from')\\\n",
    "                                            .withColumnRenamed('latitude', 'latitude_from')\\\n",
    "                                            .withColumnRenamed('stop_number', 'stop_number_from')\\\n",
    "                                            .withColumnRenamed('stop_name', 'stop_name_from')\\\n",
    "                                            .drop('elevation').drop('distance_to_Zurich')\n",
    "\n",
    "walk_to_df = df_zurich_geo.alias(\"walk_to_df\").withColumnRenamed('longitude', 'longitude_to')\\\n",
    "                                            .withColumnRenamed('latitude', 'latitude_to')\\\n",
    "                                            .withColumnRenamed('stop_number', 'stop_number_to')\\\n",
    "                                            .withColumnRenamed('stop_name', 'stop_name_to')\\\n",
    "                                            .drop('elevation').drop('distance_to_Zurich')\n",
    "\n",
    "\n",
    "df_joined = walk_from_df.crossJoin(walk_to_df)\n",
    "\n",
    "# drop the row if arrival and departure stations are the same\n",
    "mask = df_joined.stop_number_from == df_joined.stop_number_to\n",
    "df_joined = df_joined[~mask]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------------+--------------+--------------------+-----------------+\n",
      "|stop_number_from|      stop_name_from|stop_number_to|        stop_name_to|distance_as_meter|\n",
      "+----------------+--------------------+--------------+--------------------+-----------------+\n",
      "|         0000176|Zimmerberg-Basist...|       8502220|              Urdorf|        7887.3613|\n",
      "|         0000176|Zimmerberg-Basist...|       8502221|      Birmensdorf ZH|        6393.3145|\n",
      "|         0000176|Zimmerberg-Basist...|       8502222| Bonstetten-Wettswil|        4965.9766|\n",
      "|         0000176|Zimmerberg-Basist...|       8502229|   Urdorf Weihermatt|        7633.4536|\n",
      "|         0000176|Zimmerberg-Basist...|       8502559|Waldegg, Birmensd...|        4779.2036|\n",
      "+----------------+--------------------+--------------+--------------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_distances = df_joined.withColumn('distance_as_meter', calculateCrossDistance(df_joined.latitude_from,df_joined.longitude_from,df_joined.latitude_to,df_joined.longitude_to))\\\n",
    "                                .drop('longitude_from').drop('longitude_to').drop('latitude_from').drop('latitude_to')\n",
    "\n",
    "df_distances.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "create distance_dictionary -> (string, string) : float"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "distance_dictionary = {(row[0], row[1]):row[2] for row in df_distances.select('stop_number_from', 'stop_number_to', 'distance_as_meter').collect()}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Location Information of Stations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "create location_dictionary -> (string) : (string, string)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "location_dictionary = {}\n",
    "for stop_number in zurich_stations_set:\n",
    "    row = df_zurich_geo.filter(df_zurich_geo.stop_number == stop_number).select('longitude', 'latitude').collect()[0]\n",
    "    location_dictionary[stop_number] = (row[0], row[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "save data structures"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# save zurich stations set\n",
    "with open('distance/zurich_stations_set.pickle', 'wb') as handle:\n",
    "    pickle.dump(zurich_stations_set, handle)\n",
    "\n",
    "# save distance dictionary\n",
    "with open('distance/distance_dictionary.pickle', 'wb') as handle:\n",
    "    pickle.dump(distance_dictionary, handle)\n",
    "    \n",
    "# save location dictionary\n",
    "with open('distance/location_dictionary.pickle', 'wb') as handle:\n",
    "    pickle.dump(location_dictionary, handle)"
   ]
  },
  {
   "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}