ETL3_Tickets3_TaxiZone.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['PYSPARK_PYTHON'] = '/nfshome/jc9033/.conda/envs/py3/bin/python3.7'\n",
    "\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext('local', 'pyspark')\n",
    "\n",
    "from pyspark.sql.session import SparkSession\n",
    "spark = SparkSession(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://bacon.adrf.info:4044\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>pyspark</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local appName=pyspark>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "type_independent = ['1','4','6','22','25','26','29','36','59','66','78','79','82','84','85','87','88']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(0, 'Summons Number'), (1, 'Plate ID'), (2, 'Registration State'), (3, 'Plate Type'), (4, 'Issue Date'), (5, 'Violation Code'), (6, 'Vehicle Body Type'), (7, 'Vehicle Make'), (8, 'Issuing Agency'), (9, 'Street Code1'), (10, 'Street Code2'), (11, 'Street Code3'), (12, 'Vehicle Expiration Date'), (13, 'Violation Location'), (14, 'Violation Precinct'), (15, 'Issuer Precinct'), (16, 'Issuer Code'), (17, 'Issuer Command'), (18, 'Issuer Squad'), (19, 'Violation Time'), (20, 'Time First Observed'), (21, 'Violation County'), (22, 'Violation In Front Of Or Opposite'), (23, 'House Number'), (24, 'Street Name'), (25, 'Intersecting Street'), (26, 'Date First Observed'), (27, 'Law Section'), (28, 'Sub Division'), (29, 'Violation Legal Code'), (30, 'Days Parking In Effect    '), (31, 'From Hours In Effect'), (32, 'To Hours In Effect'), (33, 'Vehicle Color'), (34, 'Unregistered Vehicle?'), (35, 'Vehicle Year'), (36, 'Meter Number'), (37, 'Feet From Curb'), (38, 'Violation Post Code'), (39, 'Violation Description'), (40, 'No Standing or Stopping Violation'), (41, 'Hydrant Violation'), (42, 'Double Parking Violation')]\n",
      "CPU times: user 4 ms, sys: 4 ms, total: 8 ms\n",
      "Wall time: 2.09 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "i = \"2015\"\n",
    "path = \"data_tickets/tickets_\" + i + \".csv\"\n",
    "rdd = sc.textFile(path).cache()\n",
    "print(list(enumerate(rdd.first().split(','))))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# define functions & test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def mapper1(partitionID, rows):\n",
    "    \n",
    "    import pandas as pd\n",
    "    dict1 = pd.read_csv('output_tickets_StreetName_TaxiZone/tickets_StreetName_TaxiZone.csv').fillna(0)\\\n",
    "              .set_index('StreetName').to_dict()\n",
    "    \n",
    "    if partitionID==0:\n",
    "        next(rows)\n",
    "    import csv\n",
    "    reader = csv.reader(rows)\n",
    "    \n",
    "    for fields in reader:\n",
    "        if fields[5] not in type_independent:\n",
    "            if fields[24]:\n",
    "                try:\n",
    "                    zone = dict1['TaxiZone'][fields[24]]\n",
    "                    yield((fields[4],zone),1)\n",
    "                except:\n",
    "                    continue\n",
    "                \n",
    "def mapper2(partition):\n",
    "    for element in partition:\n",
    "        yield(element[0][0], element[0][1], element[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "i = \"2015\"\n",
    "path = \"data_tickets/tickets_\" + i + \".csv\"\n",
    "rdd = sc.textFile(path)\\\n",
    "         .mapPartitionsWithIndex(mapper1)\\\n",
    "         .groupByKey().mapValues(sum)\\\n",
    "         .mapPartitions(mapper2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 24 ms, sys: 4 ms, total: 28 ms\n",
      "Wall time: 2min 29s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[('11/20/2014', 38, 267),\n",
       " ('06/12/2015', 229, 176),\n",
       " ('05/11/2015', 163, 5),\n",
       " ('05/19/2015', 117, 96),\n",
       " ('07/01/2014', 237, 1144)]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "rdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Cleaning & Integrating"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2014\n",
      "2015\n",
      "2016\n",
      "2017\n",
      "2018\n",
      "2019\n",
      "CPU times: user 168 ms, sys: 24 ms, total: 192 ms\n",
      "Wall time: 10min 35s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "for i in range(2014,2020):\n",
    "    print(i)\n",
    "    path = \"data_tickets/tickets_\" + str(i) + \".csv\"\n",
    "    if i == 2014:\n",
    "        rdd = sc.textFile(path)\\\n",
    "                 .mapPartitionsWithIndex(mapper1)\\\n",
    "                 .groupByKey().mapValues(sum)\\\n",
    "                 .mapPartitions(mapper2)\n",
    "    else:\n",
    "        temp = sc.textFile(path)\\\n",
    "                 .mapPartitionsWithIndex(mapper1)\\\n",
    "                 .groupByKey().mapValues(sum)\\\n",
    "                 .mapPartitions(mapper2)\n",
    "        rdd = rdd.union(temp)\n",
    "    rdd.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Output CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 280 ms, sys: 84 ms, total: 364 ms\n",
      "Wall time: 3.13 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "from pyspark import SparkContext\n",
    "df = spark.createDataFrame(rdd, ['date', 'zone', 'count'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----+-----+\n",
      "|      date|zone|count|\n",
      "+----------+----+-----+\n",
      "|08/13/2000|  60|    1|\n",
      "|08/18/2000| 135|    1|\n",
      "|10/02/2000| 141|    1|\n",
      "|10/04/2000| 165|    1|\n",
      "|10/15/2000| 248|    1|\n",
      "|10/21/2000| 156|    1|\n",
      "|11/19/2000|  99|    1|\n",
      "|02/28/2001| 238|    1|\n",
      "|03/30/2001| 212|    1|\n",
      "|12/27/2001|  22|    1|\n",
      "|01/01/2004|  17|    1|\n",
      "|12/22/2005|  42|    1|\n",
      "|05/12/2010| 161|    1|\n",
      "|05/13/2010| 169|    1|\n",
      "|06/09/2011|  40|    1|\n",
      "|08/22/2011| 197|    1|\n",
      "|04/11/2012|  84|    1|\n",
      "|05/04/2012| 236|    1|\n",
      "|05/11/2012| 128|    1|\n",
      "|07/24/2012| 150|    1|\n",
      "+----------+----+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 4 ms, total: 4 ms\n",
      "Wall time: 24 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df.coalesce(1).write.format('com.databricks.spark.csv').options(header='true').save('output_tickets_zone_filterType')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Check file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "filename = ! ls output_tickets_zone_filterType | grep .csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(561439, 3)\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "pddf = pd.read_csv('output_tickets_zone_filterType/'+filename[0]).fillna(0)\n",
    "print(pddf.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "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>date</th>\n",
       "      <th>zone</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>08/13/2000</td>\n",
       "      <td>60</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>08/18/2000</td>\n",
       "      <td>135</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>10/02/2000</td>\n",
       "      <td>141</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>10/04/2000</td>\n",
       "      <td>165</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>10/15/2000</td>\n",
       "      <td>248</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "         date  zone  count\n",
       "0  08/13/2000    60      1\n",
       "1  08/18/2000   135      1\n",
       "2  10/02/2000   141      1\n",
       "3  10/04/2000   165      1\n",
       "4  10/15/2000   248      1"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pddf.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "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>date</th>\n",
       "      <th>zone</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>561434</th>\n",
       "      <td>01/02/2018</td>\n",
       "      <td>245</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>561435</th>\n",
       "      <td>01/28/2019</td>\n",
       "      <td>43</td>\n",
       "      <td>16</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>561436</th>\n",
       "      <td>01/16/2019</td>\n",
       "      <td>139</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>561437</th>\n",
       "      <td>01/28/2019</td>\n",
       "      <td>113</td>\n",
       "      <td>26</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>561438</th>\n",
       "      <td>12/30/2018</td>\n",
       "      <td>261</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "              date  zone  count\n",
       "561434  01/02/2018   245      1\n",
       "561435  01/28/2019    43     16\n",
       "561436  01/16/2019   139      1\n",
       "561437  01/28/2019   113     26\n",
       "561438  12/30/2018   261      4"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pddf.tail()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "py3",
   "language": "python",
   "name": "py3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}