preprocess_data/preprocess_data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Much of this notebook is duplicated in proprocess_data.py so that it be executed as a command-line PySpark job."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 246,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import glob\n",
    "import os\n",
    "import bz2\n",
    "import codecs\n",
    "import json\n",
    "import matplotlib.pylab as plt\n",
    "import uuid\n",
    "import random\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SQLContext, Row\n",
    "from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, DoubleType, StringType\n",
    "from pyspark.sql.functions import to_utc_timestamp, unix_timestamp, from_unixtime\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 247,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "conf = SparkConf()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 248,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.conf.SparkConf at 0x7f39a4b6fb38>"
      ]
     },
     "execution_count": 248,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf.setMaster('local[8]')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 249,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate(conf=conf)\n",
    "sql_sc = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 250,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "schema = StructType([\n",
    "        StructField('VendorID',IntegerType(),True),\n",
    "        StructField('tpep_pickup_datetime',TimestampType(),True),\n",
    "        StructField('tpep_dropoff_datetime',TimestampType(),True),\n",
    "        StructField('passenger_count',IntegerType(),True),\n",
    "        StructField('trip_distance',DoubleType(),True),\n",
    "        StructField('pickup_longitude',DoubleType(),True),\n",
    "        StructField('pickup_latitude',DoubleType(),True),\n",
    "        StructField('RateCodeID',IntegerType(),True),\n",
    "        StructField('store_and_fwd_flag',StringType(),True),\n",
    "        StructField('dropoff_longitude',DoubleType(),True),\n",
    "        StructField('dropoff_latitude',DoubleType(),True),\n",
    "        StructField('payment_type',IntegerType(),True),\n",
    "        StructField('fare_amount',DoubleType(),True),\n",
    "        StructField('extra',DoubleType(),True),\n",
    "        StructField('mta_tax',DoubleType(),True),\n",
    "        StructField('tip_amount',DoubleType(),True),\n",
    "        StructField('tolls_amount',DoubleType(),True),\n",
    "        StructField('improvement_surcharge',DoubleType(),True),\n",
    "        StructField('total_amount',DoubleType(),True),\n",
    "    ])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 251,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "infilename = 'yellow_tripdata_2015-03.csv'\n",
    "#infilename = 'sample.csv'\n",
    "raw_sdf = (sql_sc\n",
    "    .read\n",
    "    .csv(infilename, header=True, schema=schema, timestampFormat='yyyy-MM-dd HH:mm:ss'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 252,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "clean_sdf = raw_sdf.withColumn('tpep_pickup_timestamp_ms',  unix_timestamp(raw_sdf['tpep_pickup_datetime' ])*1000 + 5*60*60*1000)\n",
    "clean_sdf = clean_sdf.withColumn('tpep_dropoff_timestamp_ms', unix_timestamp(raw_sdf['tpep_dropoff_datetime'])*1000 + 5*60*60*1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 253,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 1 entries, 0 to 0\n",
      "Data columns (total 21 columns):\n",
      "VendorID                     1 non-null int64\n",
      "tpep_pickup_datetime         1 non-null datetime64[ns]\n",
      "tpep_dropoff_datetime        1 non-null datetime64[ns]\n",
      "passenger_count              1 non-null int64\n",
      "trip_distance                1 non-null float64\n",
      "pickup_longitude             1 non-null float64\n",
      "pickup_latitude              1 non-null float64\n",
      "RateCodeID                   1 non-null int64\n",
      "store_and_fwd_flag           1 non-null object\n",
      "dropoff_longitude            1 non-null float64\n",
      "dropoff_latitude             1 non-null float64\n",
      "payment_type                 1 non-null int64\n",
      "fare_amount                  1 non-null float64\n",
      "extra                        1 non-null float64\n",
      "mta_tax                      1 non-null float64\n",
      "tip_amount                   1 non-null float64\n",
      "tolls_amount                 1 non-null float64\n",
      "improvement_surcharge        1 non-null float64\n",
      "total_amount                 1 non-null float64\n",
      "tpep_pickup_timestamp_ms     1 non-null int64\n",
      "tpep_dropoff_timestamp_ms    1 non-null int64\n",
      "dtypes: datetime64[ns](2), float64(12), int64(6), object(1)\n",
      "memory usage: 248.0+ bytes\n"
     ]
    }
   ],
   "source": [
    "clean_sdf.limit(1).toPandas().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 254,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style>\n",
       "    .dataframe thead tr:only-child th {\n",
       "        text-align: right;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: left;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>VendorID</th>\n",
       "      <th>tpep_pickup_datetime</th>\n",
       "      <th>tpep_dropoff_datetime</th>\n",
       "      <th>passenger_count</th>\n",
       "      <th>trip_distance</th>\n",
       "      <th>pickup_longitude</th>\n",
       "      <th>pickup_latitude</th>\n",
       "      <th>RateCodeID</th>\n",
       "      <th>store_and_fwd_flag</th>\n",
       "      <th>dropoff_longitude</th>\n",
       "      <th>...</th>\n",
       "      <th>payment_type</th>\n",
       "      <th>fare_amount</th>\n",
       "      <th>extra</th>\n",
       "      <th>mta_tax</th>\n",
       "      <th>tip_amount</th>\n",
       "      <th>tolls_amount</th>\n",
       "      <th>improvement_surcharge</th>\n",
       "      <th>total_amount</th>\n",
       "      <th>tpep_pickup_timestamp_ms</th>\n",
       "      <th>tpep_dropoff_timestamp_ms</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-06 08:02:31</td>\n",
       "      <td>2015-03-06 08:09:55</td>\n",
       "      <td>1</td>\n",
       "      <td>1.2</td>\n",
       "      <td>-73.990211</td>\n",
       "      <td>40.750969</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.987892</td>\n",
       "      <td>...</td>\n",
       "      <td>2</td>\n",
       "      <td>7.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.3</td>\n",
       "      <td>7.80</td>\n",
       "      <td>1425646951000</td>\n",
       "      <td>1425647395000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-06 08:02:31</td>\n",
       "      <td>2015-03-06 08:15:23</td>\n",
       "      <td>1</td>\n",
       "      <td>3.2</td>\n",
       "      <td>-73.935188</td>\n",
       "      <td>40.800720</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.952553</td>\n",
       "      <td>...</td>\n",
       "      <td>2</td>\n",
       "      <td>11.5</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.3</td>\n",
       "      <td>12.30</td>\n",
       "      <td>1425646951000</td>\n",
       "      <td>1425647723000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-06 08:02:31</td>\n",
       "      <td>2015-03-06 08:12:27</td>\n",
       "      <td>1</td>\n",
       "      <td>1.1</td>\n",
       "      <td>-73.963753</td>\n",
       "      <td>40.767937</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.956947</td>\n",
       "      <td>...</td>\n",
       "      <td>2</td>\n",
       "      <td>8.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.3</td>\n",
       "      <td>8.80</td>\n",
       "      <td>1425646951000</td>\n",
       "      <td>1425647547000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-06 08:02:31</td>\n",
       "      <td>2015-03-06 08:09:09</td>\n",
       "      <td>1</td>\n",
       "      <td>0.8</td>\n",
       "      <td>-73.997177</td>\n",
       "      <td>40.742168</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-74.008064</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>6.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>1.00</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.3</td>\n",
       "      <td>7.80</td>\n",
       "      <td>1425646951000</td>\n",
       "      <td>1425647349000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-06 08:02:32</td>\n",
       "      <td>2015-03-06 08:19:37</td>\n",
       "      <td>1</td>\n",
       "      <td>2.7</td>\n",
       "      <td>-74.006844</td>\n",
       "      <td>40.730267</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.976860</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>13.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>2.75</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.3</td>\n",
       "      <td>16.55</td>\n",
       "      <td>1425646952000</td>\n",
       "      <td>1425647977000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 21 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   VendorID tpep_pickup_datetime tpep_dropoff_datetime  passenger_count  \\\n",
       "0         1  2015-03-06 08:02:31   2015-03-06 08:09:55                1   \n",
       "1         1  2015-03-06 08:02:31   2015-03-06 08:15:23                1   \n",
       "2         1  2015-03-06 08:02:31   2015-03-06 08:12:27                1   \n",
       "3         1  2015-03-06 08:02:31   2015-03-06 08:09:09                1   \n",
       "4         1  2015-03-06 08:02:32   2015-03-06 08:19:37                1   \n",
       "\n",
       "   trip_distance  pickup_longitude  pickup_latitude  RateCodeID  \\\n",
       "0            1.2        -73.990211        40.750969           1   \n",
       "1            3.2        -73.935188        40.800720           1   \n",
       "2            1.1        -73.963753        40.767937           1   \n",
       "3            0.8        -73.997177        40.742168           1   \n",
       "4            2.7        -74.006844        40.730267           1   \n",
       "\n",
       "  store_and_fwd_flag  dropoff_longitude            ...              \\\n",
       "0                  N         -73.987892            ...               \n",
       "1                  N         -73.952553            ...               \n",
       "2                  N         -73.956947            ...               \n",
       "3                  N         -74.008064            ...               \n",
       "4                  N         -73.976860            ...               \n",
       "\n",
       "   payment_type  fare_amount  extra  mta_tax  tip_amount  tolls_amount  \\\n",
       "0             2          7.0    0.0      0.5        0.00           0.0   \n",
       "1             2         11.5    0.0      0.5        0.00           0.0   \n",
       "2             2          8.0    0.0      0.5        0.00           0.0   \n",
       "3             1          6.0    0.0      0.5        1.00           0.0   \n",
       "4             1         13.0    0.0      0.5        2.75           0.0   \n",
       "\n",
       "   improvement_surcharge  total_amount  tpep_pickup_timestamp_ms  \\\n",
       "0                    0.3          7.80             1425646951000   \n",
       "1                    0.3         12.30             1425646951000   \n",
       "2                    0.3          8.80             1425646951000   \n",
       "3                    0.3          7.80             1425646951000   \n",
       "4                    0.3         16.55             1425646952000   \n",
       "\n",
       "   tpep_dropoff_timestamp_ms  \n",
       "0              1425647395000  \n",
       "1              1425647723000  \n",
       "2              1425647547000  \n",
       "3              1425647349000  \n",
       "4              1425647977000  \n",
       "\n",
       "[5 rows x 21 columns]"
      ]
     },
     "execution_count": 254,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clean_sdf.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 257,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "end_timestamp = pd.Timestamp('2015-03-03 00:00:00').tz_localize('Etc/GMT+5')\n",
    "filtered_sdf = clean_sdf.filter('tpep_dropoff_timestamp_ms <= %d' % int(end_timestamp.value / 1e6) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 258,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "771187"
      ]
     },
     "execution_count": 258,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_sdf.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 259,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style>\n",
       "    .dataframe thead tr:only-child th {\n",
       "        text-align: right;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: left;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>VendorID</th>\n",
       "      <th>tpep_pickup_datetime</th>\n",
       "      <th>tpep_dropoff_datetime</th>\n",
       "      <th>passenger_count</th>\n",
       "      <th>trip_distance</th>\n",
       "      <th>pickup_longitude</th>\n",
       "      <th>pickup_latitude</th>\n",
       "      <th>RateCodeID</th>\n",
       "      <th>store_and_fwd_flag</th>\n",
       "      <th>dropoff_longitude</th>\n",
       "      <th>...</th>\n",
       "      <th>payment_type</th>\n",
       "      <th>fare_amount</th>\n",
       "      <th>extra</th>\n",
       "      <th>mta_tax</th>\n",
       "      <th>tip_amount</th>\n",
       "      <th>tolls_amount</th>\n",
       "      <th>improvement_surcharge</th>\n",
       "      <th>total_amount</th>\n",
       "      <th>tpep_pickup_timestamp_ms</th>\n",
       "      <th>tpep_dropoff_timestamp_ms</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2</td>\n",
       "      <td>2015-03-02 11:16:24</td>\n",
       "      <td>2015-03-02 11:37:03</td>\n",
       "      <td>2</td>\n",
       "      <td>3.23</td>\n",
       "      <td>-73.987480</td>\n",
       "      <td>40.738121</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.953362</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>15.5</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>3.26</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.3</td>\n",
       "      <td>19.56</td>\n",
       "      <td>1425312984000</td>\n",
       "      <td>1425314223000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-01 16:42:33</td>\n",
       "      <td>2015-03-01 16:59:29</td>\n",
       "      <td>1</td>\n",
       "      <td>2.30</td>\n",
       "      <td>-73.979179</td>\n",
       "      <td>40.777756</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.953178</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>11.5</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>2.45</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.3</td>\n",
       "      <td>14.75</td>\n",
       "      <td>1425246153000</td>\n",
       "      <td>1425247169000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-01 16:42:33</td>\n",
       "      <td>2015-03-01 17:09:10</td>\n",
       "      <td>1</td>\n",
       "      <td>9.20</td>\n",
       "      <td>-73.865578</td>\n",
       "      <td>40.770863</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.984856</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>26.5</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>6.50</td>\n",
       "      <td>5.33</td>\n",
       "      <td>0.3</td>\n",
       "      <td>39.13</td>\n",
       "      <td>1425246153000</td>\n",
       "      <td>1425247750000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-01 16:42:34</td>\n",
       "      <td>2015-03-01 16:44:52</td>\n",
       "      <td>1</td>\n",
       "      <td>0.40</td>\n",
       "      <td>-73.952545</td>\n",
       "      <td>40.772148</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.948593</td>\n",
       "      <td>...</td>\n",
       "      <td>2</td>\n",
       "      <td>3.5</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.3</td>\n",
       "      <td>4.30</td>\n",
       "      <td>1425246154000</td>\n",
       "      <td>1425246292000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>2015-03-01 16:42:34</td>\n",
       "      <td>2015-03-01 16:47:09</td>\n",
       "      <td>1</td>\n",
       "      <td>0.90</td>\n",
       "      <td>-73.954231</td>\n",
       "      <td>40.778606</td>\n",
       "      <td>1</td>\n",
       "      <td>N</td>\n",
       "      <td>-73.947395</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>5.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.5</td>\n",
       "      <td>1.15</td>\n",
       "      <td>0.00</td>\n",
       "      <td>0.3</td>\n",
       "      <td>6.95</td>\n",
       "      <td>1425246154000</td>\n",
       "      <td>1425246429000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 21 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   VendorID tpep_pickup_datetime tpep_dropoff_datetime  passenger_count  \\\n",
       "0         2  2015-03-02 11:16:24   2015-03-02 11:37:03                2   \n",
       "1         1  2015-03-01 16:42:33   2015-03-01 16:59:29                1   \n",
       "2         1  2015-03-01 16:42:33   2015-03-01 17:09:10                1   \n",
       "3         1  2015-03-01 16:42:34   2015-03-01 16:44:52                1   \n",
       "4         1  2015-03-01 16:42:34   2015-03-01 16:47:09                1   \n",
       "\n",
       "   trip_distance  pickup_longitude  pickup_latitude  RateCodeID  \\\n",
       "0           3.23        -73.987480        40.738121           1   \n",
       "1           2.30        -73.979179        40.777756           1   \n",
       "2           9.20        -73.865578        40.770863           1   \n",
       "3           0.40        -73.952545        40.772148           1   \n",
       "4           0.90        -73.954231        40.778606           1   \n",
       "\n",
       "  store_and_fwd_flag  dropoff_longitude            ...              \\\n",
       "0                  N         -73.953362            ...               \n",
       "1                  N         -73.953178            ...               \n",
       "2                  N         -73.984856            ...               \n",
       "3                  N         -73.948593            ...               \n",
       "4                  N         -73.947395            ...               \n",
       "\n",
       "   payment_type  fare_amount  extra  mta_tax  tip_amount  tolls_amount  \\\n",
       "0             1         15.5    0.0      0.5        3.26          0.00   \n",
       "1             1         11.5    0.0      0.5        2.45          0.00   \n",
       "2             1         26.5    0.0      0.5        6.50          5.33   \n",
       "3             2          3.5    0.0      0.5        0.00          0.00   \n",
       "4             1          5.0    0.0      0.5        1.15          0.00   \n",
       "\n",
       "   improvement_surcharge  total_amount  tpep_pickup_timestamp_ms  \\\n",
       "0                    0.3         19.56             1425312984000   \n",
       "1                    0.3         14.75             1425246153000   \n",
       "2                    0.3         39.13             1425246153000   \n",
       "3                    0.3          4.30             1425246154000   \n",
       "4                    0.3          6.95             1425246154000   \n",
       "\n",
       "   tpep_dropoff_timestamp_ms  \n",
       "0              1425314223000  \n",
       "1              1425247169000  \n",
       "2              1425247750000  \n",
       "3              1425246292000  \n",
       "4              1425246429000  \n",
       "\n",
       "[5 rows x 21 columns]"
      ]
     },
     "execution_count": 259,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_sdf.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 278,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def create_events(trip_record):\n",
    "    events = []\n",
    "    trip_id = str(uuid.uuid4())\n",
    "    pickup_datetime = trip_record.tpep_pickup_timestamp_ms\n",
    "    dropoff_datetime = trip_record.tpep_dropoff_timestamp_ms\n",
    "    events.append({\n",
    "        'event_type': 'pickup',\n",
    "        'trip_id': trip_id,\n",
    "        'trip_fraction': 0.0,\n",
    "        'trip_duration_minutes': 0.0,\n",
    "        'VendorID': trip_record.VendorID,\n",
    "        'timestamp': pickup_datetime,\n",
    "        'passenger_count': trip_record.passenger_count,\n",
    "        'trip_distance': 0.0,\n",
    "        'lat': trip_record.pickup_latitude,\n",
    "        'lon': trip_record.pickup_longitude,\n",
    "        'RateCodeID': trip_record.RateCodeID,\n",
    "        'store_and_fwd_flag': trip_record.store_and_fwd_flag,\n",
    "        'payment_type': trip_record.payment_type,\n",
    "        'fare_amount': 0.0,\n",
    "        'extra': 0.0,\n",
    "        'mta_tax': 0.0,\n",
    "        'tip_amount': 0.0,\n",
    "        'tolls_amount': 0.0,\n",
    "        'improvement_surcharge': 0.0,\n",
    "        'total_amount': 0.0,\n",
    "    })\n",
    "    \n",
    "    # Create route events every 1 minute.\n",
    "    # Assume that route is a straight line from pickup to dropoff.\n",
    "    # Assume that dollar amounts are spread uniformly throughout the trip.\n",
    "    trip_duration = dropoff_datetime - pickup_datetime\n",
    "    report_period_mean = 1*60*1000\n",
    "    report_timestamp = pickup_datetime\n",
    "    while True:\n",
    "        # The next report period is a random number. This allows us to get timestamps with non-zero milliseconds.\n",
    "        report_period = random.randint(0, 2*report_period_mean)\n",
    "        report_timestamp += report_period\n",
    "        if report_timestamp >= dropoff_datetime:\n",
    "            break\n",
    "        trip_fraction = (report_timestamp - pickup_datetime) / trip_duration\n",
    "        events.append({\n",
    "            'event_type': 'route',\n",
    "            'trip_id': trip_id,\n",
    "            'trip_fraction': trip_fraction,\n",
    "            'trip_duration_minutes': trip_duration * trip_fraction / (60*1000),\n",
    "            'VendorID': trip_record.VendorID,\n",
    "            'timestamp': report_timestamp,\n",
    "            'passenger_count': trip_record.passenger_count,\n",
    "            'trip_distance': trip_record.trip_distance * trip_fraction,            \n",
    "            'lat': trip_record.pickup_latitude + (trip_record.dropoff_latitude - trip_record.pickup_latitude) * trip_fraction,\n",
    "            'lon': trip_record.pickup_longitude + (trip_record.dropoff_longitude - trip_record.pickup_longitude) * trip_fraction,\n",
    "            'RateCodeID': trip_record.RateCodeID,\n",
    "            'store_and_fwd_flag': trip_record.store_and_fwd_flag,\n",
    "            'payment_type': trip_record.payment_type,            \n",
    "            'fare_amount': trip_record.fare_amount * trip_fraction,\n",
    "            'extra': trip_record.extra * trip_fraction,\n",
    "            'mta_tax': trip_record.mta_tax * trip_fraction,\n",
    "            'tip_amount': trip_record.tip_amount * trip_fraction,\n",
    "            'tolls_amount': trip_record.tolls_amount * trip_fraction,\n",
    "            'improvement_surcharge': trip_record.improvement_surcharge * trip_fraction,\n",
    "            'total_amount': trip_record.total_amount * trip_fraction,\n",
    "        })\n",
    "\n",
    "    events.append({\n",
    "        'event_type': 'dropoff',\n",
    "        'trip_id': trip_id,\n",
    "        'trip_fraction': 1.0,\n",
    "        'trip_duration_minutes': trip_duration / (60*1000),\n",
    "        'VendorID': trip_record.VendorID,\n",
    "        'timestamp': dropoff_datetime,\n",
    "        'passenger_count': trip_record.passenger_count,\n",
    "        'trip_distance': trip_record.trip_distance,\n",
    "        'lat': trip_record.dropoff_latitude,\n",
    "        'lon': trip_record.dropoff_longitude,\n",
    "        'RateCodeID': trip_record.RateCodeID,\n",
    "        'store_and_fwd_flag': trip_record.store_and_fwd_flag,\n",
    "        'payment_type': trip_record.payment_type,\n",
    "        'fare_amount': trip_record.fare_amount,\n",
    "        'extra': trip_record.extra,\n",
    "        'mta_tax': trip_record.mta_tax,\n",
    "        'tip_amount': trip_record.tip_amount,\n",
    "        'tolls_amount': trip_record.tolls_amount,\n",
    "        'improvement_surcharge': trip_record.improvement_surcharge,\n",
    "        'total_amount': trip_record.total_amount,\n",
    "    })\n",
    "    rows = [Row(**event) for event in events]\n",
    "    return rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 279,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 0 ns, total: 0 ns\n",
      "Wall time: 183 µs\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "all_events_rdd = filtered_sdf.rdd.flatMap(create_events)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 280,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(RateCodeID=1, VendorID=2, event_type='pickup', extra=0.0, fare_amount=0.0, improvement_surcharge=0.0, lat=40.738121032714844, lon=-73.98748016357422, mta_tax=0.0, passenger_count=2, payment_type=1, store_and_fwd_flag='N', timestamp=1425312984000, tip_amount=0.0, tolls_amount=0.0, total_amount=0.0, trip_distance=0.0, trip_duration_minutes=0.0, trip_fraction=0.0, trip_id='e68f5d34-a1af-4a08-9f0f-3171549bff07'),\n",
       " Row(RateCodeID=1, VendorID=2, event_type='route', extra=0.0, fare_amount=0.7325407586763519, improvement_surcharge=0.01417820823244552, lat=40.7394738931602, lon=-73.98586769238241, mta_tax=0.023630347054075868, passenger_count=2, payment_type=1, store_and_fwd_flag='N', timestamp=1425313042556, tip_amount=0.15406986279257465, tolls_amount=0.0, total_amount=0.9244191767554479, trip_distance=0.1526520419693301, trip_duration_minutes=0.9759333333333333, trip_fraction=0.047260694108151736, trip_id='e68f5d34-a1af-4a08-9f0f-3171549bff07')]"
      ]
     },
     "execution_count": 280,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "all_events_rdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 281,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 16 ms, sys: 8 ms, total: 24 ms\n",
      "Wall time: 165 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "all_events_sdf = sql_sc.createDataFrame(all_events_rdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 282,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+-----+------------------+---------------------+------------------+------------------+--------------------+---------------+------------+------------------+-------------+-------------------+------------+------------------+-------------------+---------------------+--------------------+--------------------+\n",
      "|RateCodeID|VendorID|event_type|extra|       fare_amount|improvement_surcharge|               lat|               lon|             mta_tax|passenger_count|payment_type|store_and_fwd_flag|    timestamp|         tip_amount|tolls_amount|      total_amount|      trip_distance|trip_duration_minutes|       trip_fraction|             trip_id|\n",
      "+----------+--------+----------+-----+------------------+---------------------+------------------+------------------+--------------------+---------------+------------+------------------+-------------+-------------------+------------+------------------+-------------------+---------------------+--------------------+--------------------+\n",
      "|         1|       2|    pickup|  0.0|               0.0|                  0.0|40.738121032714844|-73.98748016357422|                 0.0|              2|           1|                 N|1425312984000|                0.0|         0.0|               0.0|                0.0|                  0.0|                 0.0|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|0.7325407586763519|  0.01417820823244552|  40.7394738931602|-73.98586769238241|0.023630347054075868|              2|           1|                 N|1425313042556|0.15406986279257465|         0.0|0.9244191767554479| 0.1526520419693301|   0.9759333333333333|0.047260694108151736|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|1.2123026634382565| 0.023463922518159805| 40.74035992017843| -73.9848116388575| 0.03910653753026634|              2|           1|                 N|1425313080906| 0.2549746246973365|         0.0|1.5298477481840191| 0.2526282324455206|               1.6151| 0.07821307506053268|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|   2.3124398708636| 0.044756900726392254| 40.74239165983901|-73.98239001312133| 0.07459483454398709|              2|           1|                 N|1425313168846| 0.4863583212267958|         0.0| 2.918149927360775| 0.4818826311541566|    3.080766666666667| 0.14918966908797418|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 3.465644874899112|  0.06707699757869248| 40.74452140540861|-73.97985157437205| 0.11179499596448748|              2|           1|                 N|1425313261028| 0.7289033736884584|         0.0|  4.37342024213075| 0.7221956739305891|    4.617133333333333| 0.22358999192897497|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 4.074861178369653|  0.07886828087167069| 40.74564650954769|-73.97851056538752| 0.13144713478611783|              2|           1|                 N|1425313309726| 0.8570353188054882|         0.0| 5.142211912832929| 0.8491484907183212|    5.428766666666666| 0.26289426957223566|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 4.557050040355126|  0.08820096852300242| 40.74653701868434|-73.97744916963616| 0.14700161420500404|              2|           1|                 N|1425313348270| 0.9584505246166263|         0.0| 5.750703147699758| 0.9496304277643262|    6.071166666666667|  0.2940032284100081|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|5.0094273607748185|  0.09695665859564163| 40.74737247169624|-73.97645339515077| 0.16159443099273607|              2|           1|                 N|1425313384431|  1.053595690072639|         0.0|6.3215741404358345|  1.043900024213075|              6.67385| 0.32318886198547214|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 6.336034705407586|  0.12263292978208232|40.749822457696084|-73.97353326258379| 0.20438821630347054|              2|           1|                 N|1425313490474| 1.3326111702986279|         0.0| 7.995667021791767| 1.3203478773204196|    8.441233333333333|  0.4087764326069411|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 7.743945117029863|  0.14988280871670703|40.752422594663116|-73.97043416543119|  0.2498046811945117|              2|           1|                 N|1425313603016| 1.6287265213882163|         0.0| 9.772359128329297| 1.6137382405165457|   10.316933333333333|  0.4996093623890234|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 8.390141242937853|  0.16238983050847458|40.753615993348895|-73.96901175633663|  0.2706497175141243|              2|           1|                 N|1425313654670| 1.7646361581920904|         0.0|10.587816949152542|  1.748397175141243|   11.177833333333334|  0.5412994350282486|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|  9.88663559322034|   0.1913542372881356|40.756379727638375| -73.9657176679191| 0.31892372881355935|              2|           1|                 N|1425313774293|  2.079382711864407|         0.0| 12.47629627118644| 2.0602472881355935|   13.171550000000002|  0.6378474576271187|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 11.21169168684423|   0.2170004842615012| 40.75882684877902|-73.96280094997107|  0.3616674737691687|              2|           1|                 N|1425313880212|   2.35807192897498|         0.0|14.148431573849878| 2.3363718805488296|   14.936866666666667|  0.7233349475383374|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|12.699641646489104|  0.24579951573849876|40.761574803239206|-73.95952566949508|  0.4096658595641646|              2|           1|                 N|1425313999152|  2.671021404358353|         0.0| 16.02612842615012| 2.6464414527845035|   16.919199999999996|  0.8193317191283292|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0| 13.17782728006457|   0.2550547215496368| 40.76245791919083| -73.9584730856636| 0.42509120258272803|              2|           1|                 N|1425314037376|  2.771594640839387|         0.0| 16.62956784503632|  2.746089168684423|   17.556266666666666|  0.8501824051654561|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|13.805058111380145|  0.26719467312348666| 40.76361629266254|-73.95709242303954| 0.44532445520581115|              2|           1|                 N|1425314087514| 2.9035154479418885|         0.0| 17.42109268765133| 2.8767959806295402|              18.3919|  0.8906489104116223|881a327e-aa56-48a...|\n",
      "|         1|       2|     route|  0.0|15.247959241323647|  0.29512179176755443|40.766281050687454|-73.95391630419961|  0.4918696529459241|              2|           1|                 N|1425314202853| 3.2069901372074248|         0.0| 19.24194082324455| 3.1774779580306696|   20.314216666666667|  0.9837393058918482|881a327e-aa56-48a...|\n",
      "|         1|       2|   dropoff|  0.0|              15.5|                  0.3|40.766746520996094|-73.95336151123047|                 0.5|              2|           1|                 N|1425314223000|               3.26|         0.0|             19.56|               3.23|                20.65|                 1.0|881a327e-aa56-48a...|\n",
      "|         1|       1|    pickup|  0.0|               0.0|                  0.0| 40.77775573730469|-73.97917938232422|                 0.0|              1|           1|                 N|1425246153000|                0.0|         0.0|               0.0|                0.0|                  0.0|                 0.0|78e42006-6ab2-429...|\n",
      "|         1|       1|     route|  0.0| 0.680050688976378| 0.017740452755905512| 40.77721050676586|-73.97764181866984| 0.02956742125984252|              1|           1|                 N|1425246213081|0.14488036417322836|         0.0|0.8722389271653543|0.13601013779527557|              1.00135| 0.05913484251968504|78e42006-6ab2-429...|\n",
      "+----------+--------+----------+-----+------------------+---------------------+------------------+------------------+--------------------+---------------+------------+------------------+-------------+-------------------+------------+------------------+-------------------+---------------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "all_events_sdf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 283,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "all_events2_sdf = all_events_sdf.withColumn('timestamp_str',  from_unixtime(all_events_sdf['timestamp' ] / 1000))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 284,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+-----+------------------+---------------------+------------------+------------------+--------------------+---------------+------------+------------------+-------------+-------------------+------------+------------------+-------------------+---------------------+--------------------+--------------------+-------------------+\n",
      "|RateCodeID|VendorID|event_type|extra|       fare_amount|improvement_surcharge|               lat|               lon|             mta_tax|passenger_count|payment_type|store_and_fwd_flag|    timestamp|         tip_amount|tolls_amount|      total_amount|      trip_distance|trip_duration_minutes|       trip_fraction|             trip_id|      timestamp_str|\n",
      "+----------+--------+----------+-----+------------------+---------------------+------------------+------------------+--------------------+---------------+------------+------------------+-------------+-------------------+------------+------------------+-------------------+---------------------+--------------------+--------------------+-------------------+\n",
      "|         1|       2|    pickup|  0.0|               0.0|                  0.0|40.738121032714844|-73.98748016357422|                 0.0|              2|           1|                 N|1425312984000|                0.0|         0.0|               0.0|                0.0|                  0.0|                 0.0|19e70b80-a85e-4e0...|2015-03-02 16:16:24|\n",
      "|         1|       2|     route|  0.0|0.7325407586763519|  0.01417820823244552|  40.7394738931602|-73.98586769238241|0.023630347054075868|              2|           1|                 N|1425313042556|0.15406986279257465|         0.0|0.9244191767554479| 0.1526520419693301|   0.9759333333333333|0.047260694108151736|19e70b80-a85e-4e0...|2015-03-02 16:17:22|\n",
      "|         1|       2|     route|  0.0|1.2123026634382565| 0.023463922518159805| 40.74035992017843| -73.9848116388575| 0.03910653753026634|              2|           1|                 N|1425313080906| 0.2549746246973365|         0.0|1.5298477481840191| 0.2526282324455206|               1.6151| 0.07821307506053268|19e70b80-a85e-4e0...|2015-03-02 16:18:00|\n",
      "|         1|       2|     route|  0.0|   2.3124398708636| 0.044756900726392254| 40.74239165983901|-73.98239001312133| 0.07459483454398709|              2|           1|                 N|1425313168846| 0.4863583212267958|         0.0| 2.918149927360775| 0.4818826311541566|    3.080766666666667| 0.14918966908797418|19e70b80-a85e-4e0...|2015-03-02 16:19:28|\n",
      "|         1|       2|     route|  0.0| 3.465644874899112|  0.06707699757869248| 40.74452140540861|-73.97985157437205| 0.11179499596448748|              2|           1|                 N|1425313261028| 0.7289033736884584|         0.0|  4.37342024213075| 0.7221956739305891|    4.617133333333333| 0.22358999192897497|19e70b80-a85e-4e0...|2015-03-02 16:21:01|\n",
      "|         1|       2|     route|  0.0| 4.074861178369653|  0.07886828087167069| 40.74564650954769|-73.97851056538752| 0.13144713478611783|              2|           1|                 N|1425313309726| 0.8570353188054882|         0.0| 5.142211912832929| 0.8491484907183212|    5.428766666666666| 0.26289426957223566|19e70b80-a85e-4e0...|2015-03-02 16:21:49|\n",
      "|         1|       2|     route|  0.0| 4.557050040355126|  0.08820096852300242| 40.74653701868434|-73.97744916963616| 0.14700161420500404|              2|           1|                 N|1425313348270| 0.9584505246166263|         0.0| 5.750703147699758| 0.9496304277643262|    6.071166666666667|  0.2940032284100081|19e70b80-a85e-4e0...|2015-03-02 16:22:28|\n",
      "|         1|       2|     route|  0.0|5.0094273607748185|  0.09695665859564163| 40.74737247169624|-73.97645339515077| 0.16159443099273607|              2|           1|                 N|1425313384431|  1.053595690072639|         0.0|6.3215741404358345|  1.043900024213075|              6.67385| 0.32318886198547214|19e70b80-a85e-4e0...|2015-03-02 16:23:04|\n",
      "|         1|       2|     route|  0.0| 6.336034705407586|  0.12263292978208232|40.749822457696084|-73.97353326258379| 0.20438821630347054|              2|           1|                 N|1425313490474| 1.3326111702986279|         0.0| 7.995667021791767| 1.3203478773204196|    8.441233333333333|  0.4087764326069411|19e70b80-a85e-4e0...|2015-03-02 16:24:50|\n",
      "|         1|       2|     route|  0.0| 7.743945117029863|  0.14988280871670703|40.752422594663116|-73.97043416543119|  0.2498046811945117|              2|           1|                 N|1425313603016| 1.6287265213882163|         0.0| 9.772359128329297| 1.6137382405165457|   10.316933333333333|  0.4996093623890234|19e70b80-a85e-4e0...|2015-03-02 16:26:43|\n",
      "|         1|       2|     route|  0.0| 8.390141242937853|  0.16238983050847458|40.753615993348895|-73.96901175633663|  0.2706497175141243|              2|           1|                 N|1425313654670| 1.7646361581920904|         0.0|10.587816949152542|  1.748397175141243|   11.177833333333334|  0.5412994350282486|19e70b80-a85e-4e0...|2015-03-02 16:27:34|\n",
      "|         1|       2|     route|  0.0|  9.88663559322034|   0.1913542372881356|40.756379727638375| -73.9657176679191| 0.31892372881355935|              2|           1|                 N|1425313774293|  2.079382711864407|         0.0| 12.47629627118644| 2.0602472881355935|   13.171550000000002|  0.6378474576271187|19e70b80-a85e-4e0...|2015-03-02 16:29:34|\n",
      "|         1|       2|     route|  0.0| 11.21169168684423|   0.2170004842615012| 40.75882684877902|-73.96280094997107|  0.3616674737691687|              2|           1|                 N|1425313880212|   2.35807192897498|         0.0|14.148431573849878| 2.3363718805488296|   14.936866666666667|  0.7233349475383374|19e70b80-a85e-4e0...|2015-03-02 16:31:20|\n",
      "|         1|       2|     route|  0.0|12.699641646489104|  0.24579951573849876|40.761574803239206|-73.95952566949508|  0.4096658595641646|              2|           1|                 N|1425313999152|  2.671021404358353|         0.0| 16.02612842615012| 2.6464414527845035|   16.919199999999996|  0.8193317191283292|19e70b80-a85e-4e0...|2015-03-02 16:33:19|\n",
      "|         1|       2|     route|  0.0| 13.17782728006457|   0.2550547215496368| 40.76245791919083| -73.9584730856636| 0.42509120258272803|              2|           1|                 N|1425314037376|  2.771594640839387|         0.0| 16.62956784503632|  2.746089168684423|   17.556266666666666|  0.8501824051654561|19e70b80-a85e-4e0...|2015-03-02 16:33:57|\n",
      "|         1|       2|     route|  0.0|13.805058111380145|  0.26719467312348666| 40.76361629266254|-73.95709242303954| 0.44532445520581115|              2|           1|                 N|1425314087514| 2.9035154479418885|         0.0| 17.42109268765133| 2.8767959806295402|              18.3919|  0.8906489104116223|19e70b80-a85e-4e0...|2015-03-02 16:34:47|\n",
      "|         1|       2|     route|  0.0|15.247959241323647|  0.29512179176755443|40.766281050687454|-73.95391630419961|  0.4918696529459241|              2|           1|                 N|1425314202853| 3.2069901372074248|         0.0| 19.24194082324455| 3.1774779580306696|   20.314216666666667|  0.9837393058918482|19e70b80-a85e-4e0...|2015-03-02 16:36:42|\n",
      "|         1|       2|   dropoff|  0.0|              15.5|                  0.3|40.766746520996094|-73.95336151123047|                 0.5|              2|           1|                 N|1425314223000|               3.26|         0.0|             19.56|               3.23|                20.65|                 1.0|19e70b80-a85e-4e0...|2015-03-02 16:37:03|\n",
      "|         1|       1|    pickup|  0.0|               0.0|                  0.0| 40.77775573730469|-73.97917938232422|                 0.0|              1|           1|                 N|1425246153000|                0.0|         0.0|               0.0|                0.0|                  0.0|                 0.0|43e77e12-1151-4a1...|2015-03-01 21:42:33|\n",
      "|         1|       1|     route|  0.0| 0.680050688976378| 0.017740452755905512| 40.77721050676586|-73.97764181866984| 0.02956742125984252|              1|           1|                 N|1425246213081|0.14488036417322836|         0.0|0.8722389271653543|0.13601013779527557|              1.00135| 0.05913484251968504|43e77e12-1151-4a1...|2015-03-01 21:43:33|\n",
      "+----------+--------+----------+-----+------------------+---------------------+------------------+------------------+--------------------+---------------+------------+------------------+-------------+-------------------+------------+------------------+-------------------+---------------------+--------------------+--------------------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "all_events2_sdf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 285,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4 ms, sys: 0 ns, total: 4 ms\n",
      "Wall time: 12.4 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "sorted_sdf = all_events2_sdf.orderBy('timestamp')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 286,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 84 ms, sys: 40 ms, total: 124 ms\n",
      "Wall time: 6min 30s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "sorted_sdf.write.mode('overwrite').format('json').save('data.json')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Playground"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}