PART II - Data Processing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dream Team:  Xingce Bao, Sohyeong Kim, Guilio Masinelli, Silvio Zanoli\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PART II. Data Processing\n",
    "\n",
    "In part II, we are computing mean and variance for all trips with same journey number within a same day(monday, tuesday, ..., sunday). Since the traffic volume is different for each day we compute mean and variance considering the days of the week. The computation is done using sparks and since **the computation takes about more than 2hours for each day**, we have saved the computed data as csv-format so that we can simply import and use these data in the next part. \n",
    "\n",
    "We also make a list of stations where direct transfer is possible, for example, if the station A have more than two lines of  bus or tram, it is considered as a station where direct transfer is possible. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### WARNING: This part takes much time to run, so please do not run unless it is necessary. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true
   },
   "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://192.168.1.123:4040\">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[8]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>streaming-silvio</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f7d855adda0>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import getpass\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "import os\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from datetime import timedelta\n",
    "\n",
    "username = getpass.getuser()\n",
    "\n",
    "spark = (SparkSession\n",
    "        .builder\n",
    "        .appName('streaming-{0}'.format(username))\n",
    "        .master('local[8]') # this number must be greater than the number of sources\n",
    "        .config('spark.executor.memory', '12g')\n",
    "        .config('spark.executor.cores', '6')\n",
    "        .getOrCreate())\n",
    "\n",
    "\n",
    "\n",
    "sc = spark.sparkContext\n",
    "conf = sc.getConf()\n",
    "\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Calculate mean and variance of the trip"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We first read the data that are preprocessed and already separted into 7 days. In this code, we are only showing the output of the processing data from 'monday' but in reality this part should be **run 7 times in total, one for each day**. The imported data has in total 13 columns and the data field is defined as:\n",
    "1. `datetime`: date of the travel (ex. 15.06.2017)\n",
    "2. `unique_id`: concatenation of the transport_id + station_id\n",
    "3. `line_id`: concatenation of the transport_type+train_number+train_type\n",
    "4. `transport_id` : transport journey which are unique for each journey\n",
    "5. `transport_type`: Bus, Tram, Zug\n",
    "6. `train_number`:  train number(ex. 1513) for Zug, operation number(ex. 85:872:3) for others\n",
    "7. `train_type`:  train type(ex. IR, IC) for train, line number(ex. 3, 700) for others\n",
    "8. `station_name`: name of the station\n",
    "9. `station_id`: BPUIC number of the station\n",
    "10. `arriving_time_scheduled`: the fixed schdule of arriving time of the train\n",
    "11. `arriving_time_actual`: the real arrving time of the train\n",
    "12. `departing_time_scheduled`: the fixed schdule of departing time of the train\n",
    "13. `departing_time_actual`: the real departing time of the train"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# First we read the data we already processed\n",
    "data_rdd_monday = sc.textFile(\"./data/monday\")\n",
    "data_rdd_monday = data_rdd_monday.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "#data_rdd_tuesday = sc.textFile(\"./data/tuesday\")\n",
    "#data_rdd_tuesday = data_rdd_tuesday.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "#data_rdd_wednesday = sc.textFile(\"./data/wednesday\")\n",
    "#data_rdd_wednesday = data_rdd_wednesday.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "#data_rdd_thursday = sc.textFile(\"./data/thursday\")\n",
    "#data_rdd_thursday = data_rdd_thursday.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "#data_rdd_friday = sc.textFile(\"./data/friday\")\n",
    "#data_rdd_friday = data_rdd_friday.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "#data_rdd_saturday = sc.textFile(\"./data/saturday\")\n",
    "#data_rdd_saturday = data_rdd_saturday.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "#data_rdd_sunday = sc.textFile(\"./data/sunday\")\n",
    "#data_rdd_sunday = data_rdd_sunday.map(lambda x : x.split(\"\\t\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['02.04.2018',\n",
       " '85:838:226851-05961-1:8587965',\n",
       " 'Bus:85:838:961:961',\n",
       " '85:838:226851-05961-1',\n",
       " 'Bus',\n",
       " '85:838:961',\n",
       " '961',\n",
       " 'Erlenbach ZH, Bahnhof',\n",
       " '8587965',\n",
       " '',\n",
       " '',\n",
       " '02.04.2018 07:33',\n",
       " '02.04.2018 07:33:00']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# See the first item in RDD\n",
    "data_rdd_monday.first()\n",
    "# data_rdd_tuesday.first()\n",
    "# data_rdd_wednesday.first()\n",
    "# data_rdd_thursday.first()\n",
    "# data_rdd_friday.first()\n",
    "# data_rdd_saturday.first()\n",
    "# data_rdd_sunday.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have notices that for some data they are missing actual arriving/departing time. These data make variances to explode. Therefore, we have filtered them out from our dataset. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['02.04.2018',\n",
       " '85:838:226851-05961-1:8587965',\n",
       " 'Bus:85:838:961:961',\n",
       " '85:838:226851-05961-1',\n",
       " 'Bus',\n",
       " '85:838:961',\n",
       " '961',\n",
       " 'Erlenbach ZH, Bahnhof',\n",
       " '8587965',\n",
       " '',\n",
       " '',\n",
       " '02.04.2018 07:33',\n",
       " '02.04.2018 07:33:00']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Filter out the unwanted data\n",
    "data_rdd_monday = data_rdd_monday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "data_rdd_monday.first()\n",
    "\n",
    "#data_rdd_tuesday = data_rdd_tuesday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "#data_rdd_tuesday.first()\n",
    "\n",
    "#data_rdd_wednesday = data_rdd_wednesday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "#data_rdd_wednesday.first()\n",
    "\n",
    "#data_rdd_thursday = data_rdd_thursday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "#data_rdd_thursday.first()\n",
    "\n",
    "#data_rdd_friday = data_rdd_friday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "#data_rdd_friday.first()\n",
    "\n",
    "#data_rdd_saturday = data_rdd_saturday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "#data_rdd_saturday.first()\n",
    "\n",
    "#data_rdd_sunday = data_rdd_sunday.filter(lambda x: not((x[12] == \"\" and x[11] != \"\") | (x[10] == \"\" and x[9] != \"\" )))\n",
    "#data_rdd_sunday.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For computation ease, we add columns which gives us the arriving/departure times in seconds for each day. We consider at '00:00' everyday as zero second and count the seconds upto the given time. For example, '02.04.2018 07:33' will be considered as '27180'seconds. We are going to call this times in second as **offset**. When there are not exists arriving/departing time(when the data is start of the travel or end of the travel), it is set to -1. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['02.04.2018',\n",
       " '85:838:226851-05961-1:8587965',\n",
       " 'Bus:85:838:961:961',\n",
       " '85:838:226851-05961-1',\n",
       " 'Bus',\n",
       " '85:838:961',\n",
       " '961',\n",
       " 'Erlenbach ZH, Bahnhof',\n",
       " '8587965',\n",
       " '',\n",
       " '',\n",
       " '02.04.2018 07:33',\n",
       " '02.04.2018 07:33:00',\n",
       " -1,\n",
       " -1,\n",
       " 27180,\n",
       " 27180]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Add columns of time as a seconds of a day\n",
    "\n",
    "# Arriving time scheduled in seconds\n",
    "data_rdd_monday = data_rdd_monday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "# Arriving time actual in seconds\n",
    "data_rdd_monday = data_rdd_monday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\\n",
    "                        (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\\n",
    "                            (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\\n",
    "                                (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))))\n",
    "                        )])\n",
    "# Departing time scheduled in seconds\n",
    "data_rdd_monday = data_rdd_monday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "# Departing time actual in seconds\n",
    "data_rdd_monday = data_rdd_monday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                        else\\\n",
    "                        (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\\n",
    "                            (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\\n",
    "                                (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))))\n",
    "                        )])\n",
    "\n",
    "# See the first item in RDD\n",
    "data_rdd_monday.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Do the same for the other days\n",
    "'''\n",
    "data_rdd_tuesday = data_rdd_tuesday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_tuesday = data_rdd_tuesday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_tuesday = data_rdd_tuesday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_tuesday = data_rdd_tuesday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_tuesday.first()\n",
    "\n",
    "\n",
    "data_rdd_wednesday = data_rdd_wednesday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_wednesday = data_rdd_wednesday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_wednesday = data_rdd_wednesday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_wednesday = data_rdd_wednesday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                       else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_wednesday.first()\n",
    "\n",
    "\n",
    "data_rdd_thursday = data_rdd_thursday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_thursday = data_rdd_thursday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_thursday = data_rdd_thursday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_thursday = data_rdd_thursday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                       else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_thursday.first()\n",
    "\n",
    "data_rdd_friday = data_rdd_friday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_friday = data_rdd_friday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_friday = data_rdd_friday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_friday = data_rdd_friday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                       else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_friday.first()\n",
    "\n",
    "data_rdd_saturday = data_rdd_saturday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_saturday = data_rdd_saturday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_saturday = data_rdd_saturday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_saturday = data_rdd_saturday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                       else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_saturday.first()\n",
    "\n",
    "data_rdd_sunday = data_rdd_sunday.map(lambda x: x + [-1 if x[9]==\"\" else int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_sunday = data_rdd_sunday.map(lambda x: x + [-1 if x[10]==\"\"\\\n",
    "                        else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[10],dayfirst=True).day==pd.to_datetime(x[9],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[10],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[10],dayfirst=True).day>pd.to_datetime(x[9],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[9],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_sunday = data_rdd_sunday.map(lambda x: x + [-1 if x[11]==\"\" else int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600))])\n",
    "data_rdd_sunday = data_rdd_sunday.map(lambda x: x + [-1 if x[12]==\"\"\\\n",
    "                       else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) if pd.to_datetime(x[12],dayfirst=True).day==pd.to_datetime(x[11],dayfirst=True).day\n",
    "                            else\\ (int(pd.to_datetime(x[12],dayfirst=True).timestamp()%(24*3600)) + 86400 if pd.to_datetime(x[12],dayfirst=True).day>pd.to_datetime(x[11],dayfirst=True).day\n",
    "                                else\\ (int(pd.to_datetime(x[11],dayfirst=True).timestamp()%(24*3600)))))])\n",
    "data_rdd_sunday.first()\n",
    "'''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we struct our data into a dataframe to use for the next steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType\n",
    "from pyspark.sql.types import StructField\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "# Define a schema for our data structure\n",
    "schema = StructType([\n",
    "    StructField(\"datetime\", StringType(), True),\n",
    "    StructField(\"unique_id\", StringType(), True),\n",
    "    StructField(\"line_id\", StringType(), True),\n",
    "    StructField(\"train_number\", StringType(), True),\n",
    "    StructField(\"transport_type\", StringType(), True),  \n",
    "    StructField(\"original_train_number\", StringType(), True),\n",
    "    StructField(\"train_type\", StringType(), True),\n",
    "    StructField(\"station_name\", StringType(), True),\n",
    "    StructField(\"station_id\", StringType(), True), \n",
    "    StructField(\"arriving_time_scheduled\", StringType(), True),\n",
    "    StructField(\"arriving_time_actual\", StringType(), True),\n",
    "    StructField(\"departing_time_scheduled\", StringType(), True),\n",
    "    StructField(\"departing_time_actual\", StringType(), True),\n",
    "    StructField(\"arrivaltimeoffsetschedule\", IntegerType(), True),\n",
    "    StructField(\"arrivaltimeoffset\", IntegerType(), True),\n",
    "    StructField(\"departuretimeoffsetschedule\", IntegerType(), True),\n",
    "    StructField(\"departuretimeoffset\", IntegerType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "data_df_monday = sqlContext.createDataFrame(data_rdd_monday, schema)\n",
    "#data_df_tuesday = sqlContext.createDataFrame(data_rdd_tuesday, schema)\n",
    "#data_df_wednesday = sqlContext.createDataFrame(data_rdd_wednesday, schema)\n",
    "#data_df_thursday = sqlContext.createDataFrame(data_rdd_thursday, schema)\n",
    "#data_df_friday = sqlContext.createDataFrame(data_rdd_friday, schema)\n",
    "#data_df_saturday = sqlContext.createDataFrame(data_rdd_saturday, schema)\n",
    "#data_df_sunday = sqlContext.createDataFrame(data_rdd_sunday, schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>datetime</th>\n",
       "      <th>unique_id</th>\n",
       "      <th>line_id</th>\n",
       "      <th>train_number</th>\n",
       "      <th>transport_type</th>\n",
       "      <th>original_train_number</th>\n",
       "      <th>train_type</th>\n",
       "      <th>station_name</th>\n",
       "      <th>station_id</th>\n",
       "      <th>arriving_time_scheduled</th>\n",
       "      <th>arriving_time_actual</th>\n",
       "      <th>departing_time_scheduled</th>\n",
       "      <th>departing_time_actual</th>\n",
       "      <th>arrivaltimeoffsetschedule</th>\n",
       "      <th>arrivaltimeoffset</th>\n",
       "      <th>departuretimeoffsetschedule</th>\n",
       "      <th>departuretimeoffset</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>02.04.2018</td>\n",
       "      <td>85:838:226851-05961-1:8587965</td>\n",
       "      <td>Bus:85:838:961:961</td>\n",
       "      <td>85:838:226851-05961-1</td>\n",
       "      <td>Bus</td>\n",
       "      <td>85:838:961</td>\n",
       "      <td>961</td>\n",
       "      <td>Erlenbach ZH, Bahnhof</td>\n",
       "      <td>8587965</td>\n",
       "      <td></td>\n",
       "      <td></td>\n",
       "      <td>02.04.2018 07:33</td>\n",
       "      <td>02.04.2018 07:33:00</td>\n",
       "      <td>-1</td>\n",
       "      <td>-1</td>\n",
       "      <td>27180</td>\n",
       "      <td>27180</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>02.04.2018</td>\n",
       "      <td>85:838:226851-05961-1:8587971</td>\n",
       "      <td>Bus:85:838:961:961</td>\n",
       "      <td>85:838:226851-05961-1</td>\n",
       "      <td>Bus</td>\n",
       "      <td>85:838:961</td>\n",
       "      <td>961</td>\n",
       "      <td>Erlenbach ZH, Im Loo</td>\n",
       "      <td>8587971</td>\n",
       "      <td>02.04.2018 07:34</td>\n",
       "      <td>02.04.2018 07:34:36</td>\n",
       "      <td>02.04.2018 07:34</td>\n",
       "      <td>02.04.2018 07:34:36</td>\n",
       "      <td>27240</td>\n",
       "      <td>27276</td>\n",
       "      <td>27240</td>\n",
       "      <td>27276</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>02.04.2018</td>\n",
       "      <td>85:838:226851-05961-1:8587964</td>\n",
       "      <td>Bus:85:838:961:961</td>\n",
       "      <td>85:838:226851-05961-1</td>\n",
       "      <td>Bus</td>\n",
       "      <td>85:838:961</td>\n",
       "      <td>961</td>\n",
       "      <td>Erlenbach ZH, Alterswohnheim</td>\n",
       "      <td>8587964</td>\n",
       "      <td>02.04.2018 07:34</td>\n",
       "      <td>02.04.2018 07:35:18</td>\n",
       "      <td>02.04.2018 07:34</td>\n",
       "      <td>02.04.2018 07:35:24</td>\n",
       "      <td>27240</td>\n",
       "      <td>27318</td>\n",
       "      <td>27240</td>\n",
       "      <td>27324</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>02.04.2018</td>\n",
       "      <td>85:838:226851-05961-1:8587973</td>\n",
       "      <td>Bus:85:838:961:961</td>\n",
       "      <td>85:838:226851-05961-1</td>\n",
       "      <td>Bus</td>\n",
       "      <td>85:838:961</td>\n",
       "      <td>961</td>\n",
       "      <td>Erlenbach ZH, Im Vogelsang</td>\n",
       "      <td>8587973</td>\n",
       "      <td>02.04.2018 07:35</td>\n",
       "      <td>02.04.2018 07:36:12</td>\n",
       "      <td>02.04.2018 07:35</td>\n",
       "      <td>02.04.2018 07:36:12</td>\n",
       "      <td>27300</td>\n",
       "      <td>27372</td>\n",
       "      <td>27300</td>\n",
       "      <td>27372</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>02.04.2018</td>\n",
       "      <td>85:838:226851-05961-1:8587976</td>\n",
       "      <td>Bus:85:838:961:961</td>\n",
       "      <td>85:838:226851-05961-1</td>\n",
       "      <td>Bus</td>\n",
       "      <td>85:838:961</td>\n",
       "      <td>961</td>\n",
       "      <td>Erlenbach ZH, Rankstrasse</td>\n",
       "      <td>8587976</td>\n",
       "      <td>02.04.2018 07:36</td>\n",
       "      <td>02.04.2018 07:37:00</td>\n",
       "      <td>02.04.2018 07:36</td>\n",
       "      <td>02.04.2018 07:37:06</td>\n",
       "      <td>27360</td>\n",
       "      <td>27420</td>\n",
       "      <td>27360</td>\n",
       "      <td>27426</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     datetime                      unique_id             line_id  \\\n",
       "0  02.04.2018  85:838:226851-05961-1:8587965  Bus:85:838:961:961   \n",
       "1  02.04.2018  85:838:226851-05961-1:8587971  Bus:85:838:961:961   \n",
       "2  02.04.2018  85:838:226851-05961-1:8587964  Bus:85:838:961:961   \n",
       "3  02.04.2018  85:838:226851-05961-1:8587973  Bus:85:838:961:961   \n",
       "4  02.04.2018  85:838:226851-05961-1:8587976  Bus:85:838:961:961   \n",
       "\n",
       "            train_number transport_type original_train_number train_type  \\\n",
       "0  85:838:226851-05961-1            Bus            85:838:961        961   \n",
       "1  85:838:226851-05961-1            Bus            85:838:961        961   \n",
       "2  85:838:226851-05961-1            Bus            85:838:961        961   \n",
       "3  85:838:226851-05961-1            Bus            85:838:961        961   \n",
       "4  85:838:226851-05961-1            Bus            85:838:961        961   \n",
       "\n",
       "                   station_name station_id arriving_time_scheduled  \\\n",
       "0         Erlenbach ZH, Bahnhof    8587965                           \n",
       "1          Erlenbach ZH, Im Loo    8587971        02.04.2018 07:34   \n",
       "2  Erlenbach ZH, Alterswohnheim    8587964        02.04.2018 07:34   \n",
       "3    Erlenbach ZH, Im Vogelsang    8587973        02.04.2018 07:35   \n",
       "4     Erlenbach ZH, Rankstrasse    8587976        02.04.2018 07:36   \n",
       "\n",
       "  arriving_time_actual departing_time_scheduled departing_time_actual  \\\n",
       "0                              02.04.2018 07:33   02.04.2018 07:33:00   \n",
       "1  02.04.2018 07:34:36         02.04.2018 07:34   02.04.2018 07:34:36   \n",
       "2  02.04.2018 07:35:18         02.04.2018 07:34   02.04.2018 07:35:24   \n",
       "3  02.04.2018 07:36:12         02.04.2018 07:35   02.04.2018 07:36:12   \n",
       "4  02.04.2018 07:37:00         02.04.2018 07:36   02.04.2018 07:37:06   \n",
       "\n",
       "   arrivaltimeoffsetschedule  arrivaltimeoffset  departuretimeoffsetschedule  \\\n",
       "0                         -1                 -1                        27180   \n",
       "1                      27240              27276                        27240   \n",
       "2                      27240              27318                        27240   \n",
       "3                      27300              27372                        27300   \n",
       "4                      27360              27420                        27360   \n",
       "\n",
       "   departuretimeoffset  \n",
       "0                27180  \n",
       "1                27276  \n",
       "2                27324  \n",
       "3                27372  \n",
       "4                27426  "
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Convert the data into Pandas Dataframe\n",
    "data_df_monday.limit(5).toPandas()\n",
    "#data_df_tuesday.limit(5).toPandas()\n",
    "#data_df_wednesday.limit(5).toPandas()\n",
    "#data_df_thursday.limit(5).toPandas()\n",
    "#data_df_friday.limit(5).toPandas()\n",
    "#data_df_saturday.limit(5).toPandas()\n",
    "#data_df_sunday.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, we are computing mean an variance of each journey(both real arrivial time and real departure time) per day and this data gives the distribution of the train schedule data. We assume that the public transport data are follwing the **gamma distribution** and also compute a shape parameter k and a scale parameter θ.\n",
    "\n",
    "There are some data having huge variance over 16 Million(4000 seconds $\\simeq$ 66.7 minutes). When the variance of the data is exploded like this, there is no way that we will choose this journey when finding the route. Therefore, we simply filtered out the data having huge variance over 16 Million. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute the mean and variance for each journey for each day and add them as additional columns\n",
    "data_mean_variance_monday = data_df_monday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "# Do the same thing for the other days\n",
    "'''\n",
    "data_mean_variance_tuesday = data_df_tuesday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "data_mean_variance_wednesday = data_df_wednesday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "data_mean_variance_thursday = data_df_thursday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "data_mean_variance_friday = data_df_friday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "data_mean_variance_saturday = data_df_saturday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "data_mean_variance_sunday = data_df_sunday.groupBy('train_number','station_id','line_id').agg(        \n",
    "    F.round(F.mean(\"arrivaltimeoffset\"),0).alias(\"avg(arrivaltimeoffset)\"),\n",
    "    F.round(F.mean(\"departuretimeoffset\"),0).alias(\"avg(departuretimeoffset)\"),\n",
    "    F.round(F.variance(\"arrivaltimeoffset\"),0).alias(\"var(arrivaltimeoffset)\"), \n",
    "    F.round(F.variance(\"departuretimeoffset\"),0).alias(\"var(departuretimeoffset)\")\n",
    ").cache()\n",
    "\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+------------------+----------------------+------------------------+----------------------+------------------------+\n",
      "|        train_number|station_id|           line_id|avg(arrivaltimeoffset)|avg(departuretimeoffset)|var(arrivaltimeoffset)|var(departuretimeoffset)|\n",
      "+--------------------+----------+------------------+----------------------+------------------------+----------------------+------------------------+\n",
      "|85:838:226864-059...|   8587964|Bus:85:838:961:961|               42631.0|                 42631.0|                 497.0|                   497.0|\n",
      "|85:849:206591-370...|   8591367| Bus:85:849:031:31|               76097.0|                 76115.0|                   NaN|                     NaN|\n",
      "|85:849:329913-370...|   8591186| Bus:85:849:031:31|                  -1.0|                 17292.0|                   NaN|                     NaN|\n",
      "|85:849:448172-370...|   8591261| Bus:85:849:031:31|               22174.0|                 22186.0|                   NaN|                     NaN|\n",
      "|85:849:554613-370...|   8591169| Bus:85:849:031:31|               26394.0|                 26406.0|                   NaN|                     NaN|\n",
      "+--------------------+----------+------------------+----------------------+------------------------+----------------------+------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check the first 5 items in the table\n",
    "data_mean_variance_monday.show(5)\n",
    "#data_mean_variance_tuesday.show(5)\n",
    "#data_mean_variance_wednesday.show(5)\n",
    "#data_mean_variance_thursday.show(5)\n",
    "#data_mean_variance_friday.show(5)\n",
    "#data_mean_variance_saturday.show(5)\n",
    "#data_mean_variance_sunday.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+------------------+----------------------+------------------------+----------------------+------------------------+\n",
      "|        train_number|station_id|           line_id|avg(arrivaltimeoffset)|avg(departuretimeoffset)|var(arrivaltimeoffset)|var(departuretimeoffset)|\n",
      "+--------------------+----------+------------------+----------------------+------------------------+----------------------+------------------------+\n",
      "|85:838:226864-059...|   8587964|Bus:85:838:961:961|               42631.0|                 42631.0|                 497.0|                   497.0|\n",
      "|85:849:144645-050...|   8591307| Bus:85:849:032:32|               61681.0|                 61693.0|                  19.0|                    19.0|\n",
      "|85:849:144669-050...|   8591291| Bus:85:849:032:32|               46964.0|                 46976.0|                 800.0|                   800.0|\n",
      "|85:849:144669-050...|   8591086| Bus:85:849:032:32|               47403.0|                 47415.0|                 420.0|                   420.0|\n",
      "|85:849:144731-050...|   8591182| Bus:85:849:032:32|               56759.0|                 56771.0|                 385.0|                   385.0|\n",
      "+--------------------+----------+------------------+----------------------+------------------------+----------------------+------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filter out the data having extremely huge variance. \n",
    "data_mean_variance_monday = data_mean_variance_monday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "data_mean_variance_monday.show(5)\n",
    "\n",
    "#data_mean_variance_tuesday = data_mean_variance_tuesday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "#data_mean_variance_tuesday.show(5)\n",
    "\n",
    "#data_mean_variance_wednesday = data_mean_variance_wednesday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "#data_mean_variance_wednesday.show(5)\n",
    "\n",
    "#data_mean_variance_thursday = data_mean_variance_thursday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "#data_mean_variance_thursday.show(5)\n",
    "\n",
    "#data_mean_variance_friday = data_mean_variance_friday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "#data_mean_variance_friday.show(5)\n",
    "\n",
    "#data_mean_variance_saturday = data_mean_variance_saturday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "#data_mean_variance_saturday.show(5)\n",
    "\n",
    "#data_mean_variance_sunday = data_mean_variance_sunday.where(col(\"var(departuretimeoffset)\")<16e6)\n",
    "#data_mean_variance_sunday.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>train_number</th>\n",
       "      <th>station_id</th>\n",
       "      <th>line_id</th>\n",
       "      <th>avg(arrivaltimeoffset)</th>\n",
       "      <th>avg(departuretimeoffset)</th>\n",
       "      <th>var(arrivaltimeoffset)</th>\n",
       "      <th>var(departuretimeoffset)</th>\n",
       "      <th>arrival_theta</th>\n",
       "      <th>departure_theta</th>\n",
       "      <th>arrival_k</th>\n",
       "      <th>departure_k</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>85:11:13794:001</td>\n",
       "      <td>8503020</td>\n",
       "      <td>Zug:13794:SN9</td>\n",
       "      <td>11504.0</td>\n",
       "      <td>11576.0</td>\n",
       "      <td>6272.0</td>\n",
       "      <td>6672.0</td>\n",
       "      <td>0.545202</td>\n",
       "      <td>0.576365</td>\n",
       "      <td>2.110045e+04</td>\n",
       "      <td>2.008450e+04</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>85:11:18:004</td>\n",
       "      <td>8503000</td>\n",
       "      <td>Zug:18:EC</td>\n",
       "      <td>60738.0</td>\n",
       "      <td>-1.0</td>\n",
       "      <td>6407.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.105486</td>\n",
       "      <td>-0.000000</td>\n",
       "      <td>5.757928e+05</td>\n",
       "      <td>-1.000000e+00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>85:11:18257:001</td>\n",
       "      <td>8503000</td>\n",
       "      <td>Zug:18257:S2</td>\n",
       "      <td>54977.0</td>\n",
       "      <td>55082.0</td>\n",
       "      <td>3447.0</td>\n",
       "      <td>2507.0</td>\n",
       "      <td>0.062699</td>\n",
       "      <td>0.045514</td>\n",
       "      <td>8.768409e+05</td>\n",
       "      <td>1.210222e+06</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>85:11:18272:002</td>\n",
       "      <td>8503202</td>\n",
       "      <td>Zug:18272:S2</td>\n",
       "      <td>68281.0</td>\n",
       "      <td>68386.0</td>\n",
       "      <td>278.0</td>\n",
       "      <td>139.0</td>\n",
       "      <td>0.004071</td>\n",
       "      <td>0.002033</td>\n",
       "      <td>1.677085e+07</td>\n",
       "      <td>3.364493e+07</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>85:11:18319:001</td>\n",
       "      <td>8503000</td>\n",
       "      <td>Zug:18319:S3</td>\n",
       "      <td>-1.0</td>\n",
       "      <td>20037.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1183.0</td>\n",
       "      <td>-0.000000</td>\n",
       "      <td>0.059041</td>\n",
       "      <td>-1.000000e+00</td>\n",
       "      <td>3.393756e+05</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "      train_number  station_id        line_id  avg(arrivaltimeoffset)  \\\n",
       "0  85:11:13794:001     8503020  Zug:13794:SN9                 11504.0   \n",
       "1     85:11:18:004     8503000      Zug:18:EC                 60738.0   \n",
       "2  85:11:18257:001     8503000   Zug:18257:S2                 54977.0   \n",
       "3  85:11:18272:002     8503202   Zug:18272:S2                 68281.0   \n",
       "4  85:11:18319:001     8503000   Zug:18319:S3                    -1.0   \n",
       "\n",
       "   avg(departuretimeoffset)  var(arrivaltimeoffset)  var(departuretimeoffset)  \\\n",
       "0                   11576.0                  6272.0                    6672.0   \n",
       "1                      -1.0                  6407.0                       0.0   \n",
       "2                   55082.0                  3447.0                    2507.0   \n",
       "3                   68386.0                   278.0                     139.0   \n",
       "4                   20037.0                     0.0                    1183.0   \n",
       "\n",
       "   arrival_theta  departure_theta     arrival_k   departure_k  \n",
       "0       0.545202         0.576365  2.110045e+04  2.008450e+04  \n",
       "1       0.105486        -0.000000  5.757928e+05 -1.000000e+00  \n",
       "2       0.062699         0.045514  8.768409e+05  1.210222e+06  \n",
       "3       0.004071         0.002033  1.677085e+07  3.364493e+07  \n",
       "4      -0.000000         0.059041 -1.000000e+00  3.393756e+05  "
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Convert the SQL query to pandas and add columns of k and theta for both arriving and departing time distribution.\n",
    "\n",
    "data_mean_variance_monday = data_mean_variance_monday.toPandas()\n",
    "data_mean_variance_monday['arrival_theta'] = data_mean_variance_monday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_monday['departure_theta'] = data_mean_variance_monday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_monday['arrival_k'] = data_mean_variance_monday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_monday['departure_k'] = data_mean_variance_monday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "\n",
    "data_mean_variance_monday.head(5)\n",
    "\n",
    "# Do the same for other days \n",
    "'''\n",
    "data_mean_variance_tuesday = data_mean_variance_tuesday.toPandas()\n",
    "data_mean_variance_tuesday['arrival_theta'] = data_mean_variance_tuesday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_tuesday['departure_theta'] = data_mean_variance_tuesday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_tuesday['arrival_k'] = data_mean_variance_tuesday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_tuesday['departure_k'] = data_mean_variance_tuesday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_tuesday.head(5)\n",
    "\n",
    "data_mean_variance_wednesday = data_mean_variance_wednesday.toPandas()\n",
    "data_mean_variance_wednesday['arrival_theta'] = data_mean_variance_wednesday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_wednesday['departure_theta'] = data_mean_variance_wednesday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_wednesday['arrival_k'] = data_mean_variance_wednesday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_wednesday['departure_k'] = data_mean_variance_wednesday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_wednesday.head(5)\n",
    "\n",
    "data_mean_variance_thursday = data_mean_variance_thursday.toPandas()\n",
    "data_mean_variance_thursday['arrival_theta'] = data_mean_variance_thursday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_thursday['departure_theta'] = data_mean_variance_thursday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_thursday['arrival_k'] = data_mean_variance_thursday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_thursday['departure_k'] = data_mean_variance_thursday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_thursday.head(5)\n",
    "\n",
    "data_mean_variance_friday = data_mean_variance_friday.toPandas()\n",
    "data_mean_variance_friday['arrival_theta'] = data_mean_variance_friday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_friday['departure_theta'] = data_mean_variance_friday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_friday['arrival_k'] = data_mean_variance_friday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_friday['departure_k'] = data_mean_variance_friday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_friday.head(5)\n",
    "\n",
    "data_mean_variance_saturday = data_mean_variance_saturday.toPandas()\n",
    "data_mean_variance_saturday['arrival_theta'] = data_mean_variance_saturday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_saturday['departure_theta'] = data_mean_variance_saturday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_saturday['arrival_k'] = data_mean_variance_saturday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_saturday['departure_k'] = data_mean_variance_saturday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_saturday.head(5)\n",
    "\n",
    "data_mean_variance_sunday = data_mean_variance_sunday.toPandas()\n",
    "data_mean_variance_sunday['arrival_theta'] = data_mean_variance_sunday[['var(arrivaltimeoffset)','avg(arrivaltimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_sunday['departure_theta'] = data_mean_variance_sunday[['var(departuretimeoffset)','avg(departuretimeoffset)']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_sunday['arrival_k'] = data_mean_variance_sunday[['avg(arrivaltimeoffset)','arrival_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_sunday['departure_k'] = data_mean_variance_sunday[['avg(departuretimeoffset)','departure_theta']].apply(lambda x: -1 if x[1]==0 or np.isnan(x[0]) else x[0]/x[1], axis=1)\n",
    "data_mean_variance_sunday.head(5)\n",
    "\n",
    "'''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, we save this processed data in csv-format so that we don't have to compute them again when we are building the routing algorithm. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Save the processed data into csv format (REAL TIME DATA)\n",
    "data_mean_variance_monday.to_csv('monday_processed.csv',index = None)\n",
    "#data_mean_variance_tuesday.to_csv('tuesday_processed.csv',index = None)\n",
    "#data_mean_variance_wednesday.to_csv('wednesday_processed.csv',index = None)\n",
    "#data_mean_variance_thursday.to_csv('thursday_processed.csv',index = None)\n",
    "#data_mean_variance_friday.to_csv('friday_processed.csv',index = None)\n",
    "#data_mean_variance_saturday.to_csv('saturday_processed.csv',index = None)\n",
    "#data_mean_variance_sunday.to_csv('sunday_processed.csv',index = None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, we also save the schedule of the travels in separate files. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Select the columns of train_number, station_id, line_id and arrivialtimeoffsetschedule and departuretimeoffsetschedule\n",
    "data_schedule_monday = data_df_monday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()\n",
    "\n",
    "#data_schedule_tuesday = data_df_tuesday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()\n",
    "#data_schedule_wednesday = data_df_wednesday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()\n",
    "#data_schedule_thursday = data_df_thursday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()\n",
    "#data_schedule_friday = data_df_friday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()\n",
    "#data_schedule_saturday = data_df_saturday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()\n",
    "#data_schedule_sunday = data_df_sunday.select('train_number','station_id','line_id','arrivaltimeoffsetschedule','departuretimeoffsetschedule').distinct().cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+-------------------+-------------------------+---------------------------+\n",
      "|        train_number|station_id|            line_id|arrivaltimeoffsetschedule|departuretimeoffsetschedule|\n",
      "+--------------------+----------+-------------------+-------------------------+---------------------------+\n",
      "|     85:11:18540:001|   8502222|       Zug:18540:S5|                    41340|                      41340|\n",
      "|     85:11:19454:001|   8502229|      Zug:19454:S14|                    54000|                      54000|\n",
      "|85:849:66828-05067-1|   8502572|  Bus:85:849:067:67|                    28260|                      28260|\n",
      "|85:3849:84063-020...|   8502572|Tram:85:3849:014:14|                    30120|                      30180|\n",
      "|85:3849:83959-020...|   8502572|Tram:85:3849:014:14|                    33720|                      33720|\n",
      "+--------------------+----------+-------------------+-------------------------+---------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Show the example of the only schedule data\n",
    "data_schedule_monday.show(5)\n",
    "#data_schedule_tuesday.show(5)\n",
    "#data_schedule_wednesday.show(5)\n",
    "#data_schedule_thursday.show(5)\n",
    "#data_schedule_friday.show(5)\n",
    "#data_schedule_saturday.show(5)\n",
    "#data_schedule_sunday.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Save the schedule data into csv-format (SCHEDULE TIME DATA)\n",
    "data_schedule_monday.toPandas().to_csv('monday_schedule.csv',index = None)\n",
    "#data_schedule_tuesday.toPandas().to_csv('tuesday_schedule.csv',index = None)\n",
    "#data_schedule_wednesday.toPandas().to_csv('wednesday_schedule.csv',index = None)\n",
    "#data_schedule_thursday.toPandas().to_csv('thursday_schedule.csv',index = None)\n",
    "#data_schedule_friday.toPandas().to_csv('friday_schedule.csv',index = None)\n",
    "#data_schedule_saturday.toPandas().to_csv('saturday_schedule.csv',index = None)\n",
    "#data_schedule_sunday.toPandas().to_csv('sunday_schedule.csv',index = None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ----   Calculating mean and variance is finished at this point. \n",
    "**(If you don't have processed data, you have to run this part after uncommenting all the lines for all days. The estimated running time to process the all data is around 16 hours)**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Find a stations where direct transfer is possible\n",
    "\n",
    "Using the transfer_station data created from the Part I, we sort out the station where we can make a direct transfer. It is same as finding stations where there are more than two lines which is equvalent of having more than one `transfer_count`. The sorted list is then save as a pickle. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pickle\n",
    "\n",
    "# First we read the data we already processed\n",
    "transfer_stations = sc.textFile(\"./data/transfer_stations\")\n",
    "transfer_stations = transfer_stations.map(lambda x : x.split(\"\\t\"))\n",
    "\n",
    "# Choose the station which has more than one line of the transportation\n",
    "transfer_stations = transfer_stations.filter(lambda x : int(x[1]) > 1)\n",
    "transfer_station_list = [int(i[0]) for i in transfer_stations.collect()]\n",
    "\n",
    "# Save it as pickle\n",
    "pickle.dump(transfer_station_list, open( \"transfer.p\", \"wb\" ) )"
   ]
  }
 ],
 "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
}