cohort2/exploration/wiggles_spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import matplotlib.pyplot as plt\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import time\n",
    "from os import listdir\n",
    "from os.path import isfile, join\n",
    "from datetime import time\n",
    "import glob\n",
    "import sys\n",
    "import datetime\n",
    "\n",
    "from pyspark.sql.functions import udf, hour, mean, minute, stddev, count, max as psmax, min as psmin, date_format\n",
    "\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Go here to see pyspark functions\n",
    "http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#data locations\n",
    "data_5min_path_d11 = \"../../../d11_traffic_data/d11/\"\n",
    "data_5min_path_d8 = \"../../../d8_traffic_data/\"\n",
    "meta_path_d11 = \"../../../d11_traffic_data/meta/d11/\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Get data file names"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These are all of the filenames that are going to be fed into the Spark instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#get all files to process\n",
    "#onlyfiles = [f for f in listdir(data_5min_path) if isfile(join(data_5min_path, f)) ]\n",
    "#onlyfiles = [a for a in onlyfiles if 'txt.gz' in a]\n",
    "onlyfiles = [data_5min_path_d11 + a for a in listdir(data_5min_path_d11) if 'txt.gz' in a ] + \\\n",
    "            [data_5min_path_d8 + b for b in listdir(data_5min_path_d8) if 'txt.gz' in b ] \n",
    "onlyfiles"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Make spark schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "struct list was generated with the following code after reading the files with inferschema = true, then hand modified \n",
    "```\n",
    "'[' + ','.join(['StructField(\"%s\",%s(),True)'% (colnames[idx], str(i.dataType))\n",
    "for idx, i in enumerate(rdd.schema)]) + ']'\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " Column Names are not defined in the files. Defining column names for readability"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "colnames = [\n",
    "    'Timestamp','Station','District','Freeway','Direction_of_Travel',\n",
    "    'LaneType','StationLength','Samples',\n",
    "    'Perc_Observed','TotalFlow','AvgOccupancy','AvgSpeed',\n",
    "    'Lane1_Samples','Lane1_Flow','Lane1_AvgOcc','Lane1_AvgSpeed','Lane1_Observed',\n",
    "    'Lane2_Samples','Lane2_Flow','Lane2_AvgOcc','Lane2_AvgSpeed','Lane2_Observed',\n",
    "    'Lane3_Samples','Lane3_Flow','Lane3_AvgOcc','Lane3_AvgSpeed','Lane3_Observed',\n",
    "    'Lane4_Samples','Lane4_Flow','Lane4_AvgOcc','Lane4_AvgSpeed','Lane4_Observed',\n",
    "    'Lane5_Samples','Lane5_Flow','Lane5_AvgOcc','Lane5_AvgSpeed','Lane5_Observed',\n",
    "    'Lane6_Samples','Lane6_Flow','Lane6_AvgOcc','Lane6_AvgSpeed','Lane6_Observed',\n",
    "    'Lane7_Samples','Lane7_Flow','Lane7_AvgOcc','Lane7_AvgSpeed','Lane7_Observed',\n",
    "    'Lane8_Samples','Lane8_Flow','Lane8_AvgOcc','Lane8_AvgSpeed','Lane8_Observed'\n",
    "]\n",
    "colnames = [c.lower() for c in colnames]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ran spark instance once with limited dataset to identify the datatypes of each column to help create the following script."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#print '[\\n    ' + \",\\n    \".join(['StructField(\"%s\",%s(),True)'% (colnames[idx], str(i.dataType))\n",
    "#for idx, i in enumerate(rdd.schema)]) + '\\n]'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Build dataframe with spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Defining the schema of the files that we are reading in. StructType creates the dataframe schema equivalent of create syntax in SQL)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "struct_list = [\n",
    "    StructField(\"timestamp\",TimestampType(),True),\n",
    "    StructField(\"station\",IntegerType(),True),\n",
    "    StructField(\"district\",IntegerType(),True),\n",
    "    StructField(\"freeway\",IntegerType(),True),\n",
    "    StructField(\"direction_of_travel\",StringType(),True),\n",
    "    StructField(\"lanetype\",StringType(),True),\n",
    "    StructField(\"stationlength\",DoubleType(),True),\n",
    "    StructField(\"samples\",IntegerType(),True),\n",
    "    StructField(\"perc_observed\",IntegerType(),True),\n",
    "    StructField(\"totalflow\",IntegerType(),True),\n",
    "    StructField(\"avgoccupancy\",DoubleType(),True),\n",
    "    StructField(\"avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane1_samples\",IntegerType(),True),\n",
    "    StructField(\"lane1_flow\",IntegerType(),True),\n",
    "    StructField(\"lane1_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane1_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane1_observed\",IntegerType(),True),\n",
    "    StructField(\"lane2_samples\",IntegerType(),True),\n",
    "    StructField(\"lane2_flow\",IntegerType(),True),\n",
    "    StructField(\"lane2_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane2_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane2_observed\",IntegerType(),True),\n",
    "    StructField(\"lane3_samples\",IntegerType(),True),\n",
    "    StructField(\"lane3_flow\",IntegerType(),True),\n",
    "    StructField(\"lane3_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane3_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane3_observed\",IntegerType(),True),\n",
    "    StructField(\"lane4_samples\",IntegerType(),True),\n",
    "    StructField(\"lane4_flow\",IntegerType(),True),\n",
    "    StructField(\"lane4_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane4_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane4_observed\",IntegerType(),True),\n",
    "    StructField(\"lane5_samples\",IntegerType(),True),\n",
    "    StructField(\"lane5_flow\",IntegerType(),True),\n",
    "    StructField(\"lane5_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane5_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane5_observed\",IntegerType(),True),\n",
    "    StructField(\"lane6_samples\",IntegerType(),True),\n",
    "    StructField(\"lane6_flow\",IntegerType(),True),\n",
    "    StructField(\"lane6_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane6_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane6_observed\",IntegerType(),True),\n",
    "    StructField(\"lane7_samples\",IntegerType(),True),\n",
    "    StructField(\"lane7_flow\",IntegerType(),True),\n",
    "    StructField(\"lane7_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane7_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane7_observed\",IntegerType(),True),\n",
    "    StructField(\"lane8_samples\",IntegerType(),True),\n",
    "    StructField(\"lane8_flow\",IntegerType(),True),\n",
    "    StructField(\"lane8_avgocc\",DoubleType(),True),\n",
    "    StructField(\"lane8_avgspeed\",DoubleType(),True),\n",
    "    StructField(\"lane8_observed\",IntegerType(),True)\n",
    "]\n",
    "\n",
    "schema_struct = StructType(struct_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Loading the data into spark dataframe from the files (equivalent of insert statements with files in SQL)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "limit_files = onlyfiles[:5] + onlyfiles[-5:]\n",
    "limit_files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#node this is only the first 5 days of files for now\n",
    "limit_files = onlyfiles[:5] + onlyfiles[-5:]\n",
    "files = [filename for filename in limit_files]\n",
    "\n",
    "rdd = spark.read.csv(\n",
    "    files, \n",
    "    header='false',\n",
    "    timestampFormat='MM/dd/yyyy HH:mm:ss',\n",
    "    schema=schema_struct,\n",
    "    inferSchema='false'\n",
    ")\n",
    "    \n",
    "rdd.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build freeway STM station order from meta data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Build a sparksql dataframe with the metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "meta_path = \"../../../d11_traffic_data/meta/d11/\"\n",
    "\n",
    "def loadMeta():\n",
    "    meta_dir= meta_path+'d11_text_meta_2015_*.txt'\n",
    "    meta_files = glob.glob(meta_dir)\n",
    "\n",
    "    meta_file_list = []\n",
    "    for meta_file in meta_files:\n",
    "        date = str('_'.join(meta_file.split('_')[4:7])).split('.')[0]\n",
    "        df = pd.read_table(meta_file, index_col=None, header=0)\n",
    "        date_col = pd.Series([date] * len(df))\n",
    "        df['file_date'] = date_col\n",
    "        # drop rows that are missing latitude / longitude values\n",
    "        #df.dropna(inplace=True, subset=['Latitude', 'Longitude'], how='any')\n",
    "        meta_file_list.append(df)\n",
    "\n",
    "    meta_frame = pd.concat(meta_file_list).drop_duplicates(subset='ID', keep='last')\n",
    "\n",
    "    usefwy = [ 56, 125, 805,  52, 163,   8,  15,   5, 905,  78,  94,  54]\n",
    "\n",
    "    meta_frame = meta_frame[meta_frame.Fwy.apply(lambda x: x in usefwy)]\n",
    "\n",
    "    #Add freeway name FwyDir\n",
    "    meta_frame['freeway'] = meta_frame.Fwy.apply(str) + meta_frame.Dir\n",
    "    \n",
    "    r_c = {}\n",
    "    for c in meta_frame.columns:\n",
    "        r_c[c]=c.lower()\n",
    "    \n",
    "    meta_frame=meta_frame.rename(columns = r_c )\n",
    "    return meta_frame\n",
    "\n",
    "meta_data = sqlCtx.createDataFrame(loadMeta().loc[:,['id','abs_pm','type']].rename(columns={'id':'station'}))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "meta_data.show(100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# filter for weekdays I5 S\n",
    "# group by station, time\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Modify this to make all queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "weekdaySelector = udf(\n",
    "    lambda x: \"weekday\" if int(x) < 6 else \"weekend\"\n",
    ")\n",
    "\n",
    "timeOfDay = udf(\n",
    "    lambda x: time(int(x.hour), int(x.minute)).strftime(\"%H:%M\")\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "station_time = (\n",
    "    rdd\n",
    "    .select(\n",
    "        'district',\n",
    "        'freeway',\n",
    "        'direction_of_travel',\n",
    "        'timestamp',\n",
    "        'station',\n",
    "        'totalflow',\n",
    "        'avgoccupancy',\n",
    "        'avgspeed',\n",
    "        date_format('timestamp', 'u').alias('dayofweek')\n",
    "    )\n",
    "    \n",
    ")\n",
    "\n",
    "station_time = (\n",
    "    station_time\n",
    "    .withColumn(\n",
    "        'dayType', \n",
    "        weekdaySelector(station_time.dayofweek)\n",
    "        )\n",
    "    .withColumn(\n",
    "        'timeOfDay',\n",
    "        timeOfDay(station_time.timestamp)\n",
    "    )\n",
    "    .groupBy([\n",
    "        'district',\n",
    "        'freeway',\n",
    "        'direction_of_travel',\n",
    "        'station',\n",
    "        'dayType',\n",
    "        'timeOfDay'\n",
    "    ])\n",
    "    .agg(\n",
    "        mean(\"totalflow\").alias(\"flow_mean\"),\n",
    "        stddev(\"totalflow\").alias(\"flow_std\"),\n",
    "        count(\"totalflow\").alias(\"flow_count\"),\n",
    "        psmax(\"totalflow\").alias(\"flow_max\"),\n",
    "        psmin(\"totalflow\").alias(\"flow_min\"),\n",
    "        \n",
    "        mean(\"avgoccupancy\").alias(\"occ_mean\"),\n",
    "        stddev(\"avgoccupancy\").alias(\"occ_std\"),\n",
    "        count(\"avgoccupancy\").alias(\"occ_count\"),\n",
    "        psmax(\"avgoccupancy\").alias(\"occ_max\"),\n",
    "        psmin(\"avgoccupancy\").alias(\"occ_min\"),\n",
    "        \n",
    "        mean(\"avgspeed\").alias(\"speed_mean\"),\n",
    "        stddev(\"avgspeed\").alias(\"speed_std\"),\n",
    "        count(\"avgspeed\").alias(\"speed_count\"),\n",
    "        psmax(\"avgspeed\").alias(\"speed_max\"),\n",
    "        psmin(\"avgspeed\").alias(\"speed_min\")\n",
    "    )\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df = station_time.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df.sort_values('timeOfDay')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def reduce_data_by_dict(df, keyval_dict):\n",
    "    for key, val in keyval_dict.iteritems():\n",
    "        df = df[df[key] == val]\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%run ../trafficpassion/AnalyzeWiggles.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "restrict_dict = {'dayType': 'weekday', \n",
    "                 'district': 11,\n",
    "                 'freeway': 5,\n",
    "                 'direction_of_travel': 'S'\n",
    "                }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "reduce_data_by_dict(df,restrict_dict ).sort_values(['timeOfDay'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}