preprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "e6885863",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "21/10/19 22:11:52 WARN Utils: Your hostname, Sumans-MacBook-Pro.local resolves to a loopback address: 127.0.0.1; using 192.168.0.102 instead (on interface en0)\n",
      "21/10/19 22:11:52 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "21/10/19 22:11:52 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\n",
      "      ____              __\n",
      "     / __/__  ___ _____/ /__\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 3.0.3\n",
      "      /_/\n",
      "\n",
      "Using Python version 3.8.2 (default, Apr  8 2021 23:19:18)\n",
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "exec(open(os.path.join(os.environ[\"SPARK_HOME\"], 'python/pyspark/shell.py')).read())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5e664710",
   "metadata": {},
   "source": [
    "#### Add the config file within the spark context"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "9d8c54cd",
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.addFile('/Users/sumangangopadhyay/complex-spark-transformations/config.py')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fb0ded59",
   "metadata": {},
   "source": [
    "#### Import the relevant libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d631ca35",
   "metadata": {},
   "outputs": [],
   "source": [
    "import config as cf\n",
    "from pyspark.sql import functions as func\n",
    "from pyspark.sql.types import IntegerType"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "820ec4ad",
   "metadata": {},
   "source": [
    "#### Get the relevant configurations in variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "b52f54c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_path = cf.data_path()\n",
    "clean_data_path = cf.clean_data_path()\n",
    "primary_response_variables = cf.primary_response_variables().split(',')\n",
    "secondary_response_variables = cf.secondary_response_variables().split(',')\n",
    "primary_explanatory_variables = cf.primary_explanatory_variables().split(',')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f67fa78d",
   "metadata": {},
   "source": [
    "#### Read the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "25330ac1",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(data_path, header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "be79c4bf",
   "metadata": {},
   "source": [
    "#### Remove the spaces from the column names so that it's easier to use the columns later on"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "61af4c58",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_with_no_spaces_in_colm_names = df.select([func.col(col).alias(col.replace(' ', '_')) for col in df.columns])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cb258156",
   "metadata": {},
   "source": [
    "#### Get rid of the initial dataframe to free up memory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "8773033e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Summons Number: string, Plate ID: string, Registration State: string, Plate Type: string, Issue Date: string, Violation Code: string, Vehicle Body Type: string, Vehicle Make: string, Issuing Agency: string, Street Code1: string, Street Code2: string, Street Code3: string, Vehicle Expiration Date: string, Violation Location: string, Violation Precinct: string, Issuer Precinct: string, Issuer Code: string, Issuer Command: string, Issuer Squad: string, Violation Time: string, Time First Observed: string, Violation County: string, Violation In Front Of Or Opposite: string, House Number: string, Street Name: string, Intersecting Street: string, Date First Observed: string, Law Section: string, Sub Division: string, Violation Legal Code: string, Days Parking In Effect    : string, From Hours In Effect: string, To Hours In Effect: string, Vehicle Color: string, Unregistered Vehicle?: string, Vehicle Year: string, Meter Number: string, Feet From Curb: string, Violation Post Code: string, Violation Description: string, No Standing or Stopping Violation: string, Hydrant Violation: string, Double Parking Violation: string]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.unpersist()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "baa0c3dd",
   "metadata": {},
   "source": [
    "#### Get the count of distinct values of the attributes which form the response variables ( In statistical terms, response variables are the variables on the y-axis, i.e. the variables whose variations are being observed)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "309e2a73",
   "metadata": {},
   "outputs": [],
   "source": [
    "unique_count_of_primary_response_variables = df_with_no_spaces_in_colm_names\\\n",
    ".select([func.countDistinct(col)\\\n",
    "         .alias('unique_'+ col) for col in primary_response_variables])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "530cd9cc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "21/10/19 22:12:42 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n",
      "[Stage 1:===================================>                     (10 + 6) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+-----------------+---------------------+------------------+---------------------------+---------------------+\n",
      "|unique_Registration_State|unique_Plate_Type|unique_Violation_Code|unique_Law_Section|unique_Violation_Legal_Code|unique_Issuing_Agency|\n",
      "+-------------------------+-----------------+---------------------+------------------+---------------------------+---------------------+\n",
      "|                       67|               86|                  100|                 8|                          4|                   17|\n",
      "+-------------------------+-----------------+---------------------+------------------+---------------------------+---------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "unique_count_of_primary_response_variables.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "0ec764c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "unique_count_of_secondary_response_variables = df_with_no_spaces_in_colm_names\\\n",
    ".select([func.countDistinct(col)\\\n",
    "         .alias('unique_'+ col) for col in secondary_response_variables])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "410cbc69",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 4:===>                                                     (1 + 15) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------+-------------------+-------------------+\n",
      "|unique_Violation_County|unique_Issuer_Squad|unique_Vehicle_Year|\n",
      "+-----------------------+-------------------+-------------------+\n",
      "|                     12|                 49|                100|\n",
      "+-----------------------+-------------------+-------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "unique_count_of_secondary_response_variables.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1bda2328",
   "metadata": {},
   "source": [
    "#### Check the NaNs and Nulls in the explanatory variables. These variables typically go in the x-axis. Statistically, we are interested in the extent to which the variation in the response variables are associated with the variation in these variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "33c05cce",
   "metadata": {},
   "outputs": [],
   "source": [
    "nan_null_count_in_primary_explanatory_variables = df_with_no_spaces_in_colm_names\\\n",
    ".select([func.count(func.when(func.isnan(col) | func.col(col).isNull(), col))\\\n",
    "         .alias('null_nan_count_'+ col) for col in primary_explanatory_variables])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "55f618d4",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 7:==============================================>          (13 + 3) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+-----------------------------+\n",
      "|null_nan_count_Issue_Date|null_nan_count_Violation_Time|\n",
      "+-------------------------+-----------------------------+\n",
      "|                        0|                           63|\n",
      "+-------------------------+-----------------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 7:=====================================================>   (15 + 1) / 16]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "nan_null_count_in_primary_explanatory_variables.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9046ee7",
   "metadata": {},
   "source": [
    "#### Creating a new categorical explanatory variable (Categorical variables are factors with 2 or more levels, e.g. a rainbow is a factor with 7 levels)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "d301c9e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".withColumn('Violation_AM_or_PM', \\\n",
    "            func.when(func.isnan(df_with_no_spaces_in_colm_names.Violation_Time) \\\n",
    "                      | func.col('Violation_Time').isNull()\\\n",
    "                      , func.lit(None))\\\n",
    "            .otherwise(func.substring(df_with_no_spaces_in_colm_names.Violation_Time,5,1)))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5dc34071",
   "metadata": {},
   "source": [
    "#### Verify that the new column has been populated correctly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "4e193e2a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------------+\n",
      "|Violation_Time|Violation_AM_or_PM|\n",
      "+--------------+------------------+\n",
      "|         0143A|                 A|\n",
      "|         0400P|                 P|\n",
      "|         0233P|                 P|\n",
      "|         1120A|                 A|\n",
      "|         0555P|                 P|\n",
      "|         0852P|                 P|\n",
      "|         0215A|                 A|\n",
      "|         0758A|                 A|\n",
      "|         1005A|                 A|\n",
      "|         0845A|                 A|\n",
      "+--------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select('Violation_Time','Violation_AM_or_PM')\\\n",
    ".filter(df_with_no_spaces_in_colm_names.Violation_Time.isNotNull())\\\n",
    ".show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "8663eb57",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 11:>                                                         (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------------+\n",
      "|Violation_Time|Violation_AM_or_PM|\n",
      "+--------------+------------------+\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "|          null|              null|\n",
      "+--------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select('Violation_Time','Violation_AM_or_PM')\\\n",
    ".filter(df_with_no_spaces_in_colm_names.Violation_Time.isNull())\\\n",
    ".show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "445636b9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 14:=====>                                                  (1 + 10) / 11]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+------------------+\n",
      "|Violation_Time|Violation_AM_or_PM|\n",
      "+--------------+------------------+\n",
      "+--------------+------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 14:=========================================>               (8 + 3) / 11]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select('Violation_Time','Violation_AM_or_PM')\\\n",
    ".filter(func.isnan(df_with_no_spaces_in_colm_names.Violation_Time))\\\n",
    ".show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1ceca0c7",
   "metadata": {},
   "source": [
    "#### Convert the dates to a standard YYYY-MM-DD format so that it becomes easy to extract the weeks, months and years from it"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "8a970637",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".withColumn('Issue_Date_standardized', \\\n",
    "            func.date_format(func.unix_timestamp('Issue_Date','MM/dd/yyyy').cast('timestamp'), 'yyyy-MM-dd'))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bc3614d3",
   "metadata": {},
   "source": [
    "#### Now drop the original column and rename this new column as the original one"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "9328309a",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".drop('Issue_Date')\\\n",
    ".withColumnRenamed('Issue_Date_standardized','Issue_Date')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "beb80a2b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|Issue_Date|\n",
      "+----------+\n",
      "|2016-07-10|\n",
      "|2016-07-08|\n",
      "|2016-08-23|\n",
      "|2017-06-14|\n",
      "|2016-11-21|\n",
      "|2017-06-13|\n",
      "|2016-08-03|\n",
      "|2016-12-21|\n",
      "|2016-11-21|\n",
      "|2016-10-05|\n",
      "+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names.select('Issue_Date').show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "da89aaba",
   "metadata": {},
   "source": [
    "#### Now let's create some more categorical variables from the Issue Date column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "40ab8fe8",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".withColumn('Issue_Year', func.year('Issue_Date'))\n",
    "\n",
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".withColumn('Issue_Month', func.month('Issue_Date'))\n",
    "\n",
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".withColumn('Issue_DayofWeek', func.dayofweek('Issue_Date'))\n",
    "\n",
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".withColumn('Issue_DayofMonth', func.dayofmonth('Issue_Date'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "7f1ed8da",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+-----------+---------------+----------------+\n",
      "|Issue_Date|Issue_Year|Issue_Month|Issue_DayofWeek|Issue_DayofMonth|\n",
      "+----------+----------+-----------+---------------+----------------+\n",
      "|2016-07-10|      2016|          7|              1|              10|\n",
      "|2016-07-08|      2016|          7|              6|               8|\n",
      "|2016-08-23|      2016|          8|              3|              23|\n",
      "|2017-06-14|      2017|          6|              4|              14|\n",
      "|2016-11-21|      2016|         11|              2|              21|\n",
      "|2017-06-13|      2017|          6|              3|              13|\n",
      "|2016-08-03|      2016|          8|              4|               3|\n",
      "|2016-12-21|      2016|         12|              4|              21|\n",
      "|2016-11-21|      2016|         11|              2|              21|\n",
      "|2016-10-05|      2016|         10|              4|               5|\n",
      "+----------+----------+-----------+---------------+----------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names.select('Issue_Date','Issue_Year',\\\n",
    "                                       'Issue_Month','Issue_DayofWeek','Issue_DayofMonth').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "3385dc07",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------+--------------------+------------------+\n",
      "|Days_Parking_In_Effect____|From_Hours_In_Effect|To_Hours_In_Effect|\n",
      "+--------------------------+--------------------+------------------+\n",
      "|                      null|                null|              null|\n",
      "|                      null|                null|              null|\n",
      "|                      null|                null|              null|\n",
      "|                         Y|               0700A|             0700P|\n",
      "|                         Y|               0700A|             0700P|\n",
      "|                      null|                null|              null|\n",
      "|                   BBBBBBB|                 ALL|               ALL|\n",
      "|                      null|                null|              null|\n",
      "|                      null|                null|              null|\n",
      "|                      null|                null|              null|\n",
      "+--------------------------+--------------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names.select('Days_Parking_In_Effect____','From_Hours_In_Effect','To_Hours_In_Effect').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "b292e948",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 18:===>                                                    (1 + 15) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------+-----------------------------------+---------------------------------+\n",
      "|null_nan_count_Days_Parking_In_Effect____|null_nan_count_From_Hours_In_Effect|null_nan_count_To_Hours_In_Effect|\n",
      "+-----------------------------------------+-----------------------------------+---------------------------------+\n",
      "|                                  2712416|                            5450946|                          5450943|\n",
      "+-----------------------------------------+-----------------------------------+---------------------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select([func.count(func.when(func.isnan(col) | func.col(col).isNull(), col))\\\n",
    "         .alias('null_nan_count_'+ col) for col in \\\n",
    "        ['Days_Parking_In_Effect____','From_Hours_In_Effect','To_Hours_In_Effect']]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "de24c6bd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------------+---------------------+--------------+\n",
      "|Violation_In_Front_Of_Or_Opposite|Unregistered_Vehicle?|Feet_From_Curb|\n",
      "+---------------------------------+---------------------+--------------+\n",
      "|                             null|                 null|             0|\n",
      "|                             null|                 null|             0|\n",
      "|                             null|                 null|             0|\n",
      "|                                O|                 null|             0|\n",
      "|                                F|                 null|             0|\n",
      "|                             null|                 null|             0|\n",
      "|                                F|                    0|             1|\n",
      "|                             null|                 null|             0|\n",
      "|                             null|                 null|             0|\n",
      "|                             null|                 null|             0|\n",
      "+---------------------------------+---------------------+--------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select('Violation_In_Front_Of_Or_Opposite','Unregistered_Vehicle?','Feet_From_Curb').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "fab9142a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 21:===>                                                    (1 + 15) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------------+------------------------------------+-----------------------------+\n",
      "|null_nan_count_Violation_In_Front_Of_Or_Opposite|null_nan_count_Unregistered_Vehicle?|null_nan_count_Feet_From_Curb|\n",
      "+------------------------------------------------+------------------------------------+-----------------------------+\n",
      "|                                         2161235|                             9675432|                            0|\n",
      "+------------------------------------------------+------------------------------------+-----------------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select([func.count(func.when(func.isnan(col) | func.col(col).isNull(), col))\\\n",
    "         .alias('null_nan_count_'+ col) for col in \\\n",
    "        ['Violation_In_Front_Of_Or_Opposite','Unregistered_Vehicle?','Feet_From_Curb']]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "5ac19e6d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------------+-----------------+------------------------+\n",
      "|No_Standing_or_Stopping_Violation|Hydrant_Violation|Double_Parking_Violation|\n",
      "+---------------------------------+-----------------+------------------------+\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "|                             null|             null|                    null|\n",
      "+---------------------------------+-----------------+------------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select('No_Standing_or_Stopping_Violation','Hydrant_Violation','Double_Parking_Violation').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "289f8e04",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 24:>                                                       (0 + 16) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------------+--------------------------------+---------------------------------------+\n",
      "|null_nan_count_No_Standing_or_Stopping_Violation|null_nan_count_Hydrant_Violation|null_nan_count_Double_Parking_Violation|\n",
      "+------------------------------------------------+--------------------------------+---------------------------------------+\n",
      "|                                        10803028|                        10803028|                               10803028|\n",
      "+------------------------------------------------+--------------------------------+---------------------------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 24:===>                                                    (1 + 15) / 16]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names\\\n",
    ".select([func.count(func.when(func.isnan(col) | func.col(col).isNull(), col))\\\n",
    "         .alias('null_nan_count_'+ col) for col in \\\n",
    "        ['No_Standing_or_Stopping_Violation','Hydrant_Violation','Double_Parking_Violation']]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "38f74e28",
   "metadata": {},
   "source": [
    "#### Based on the information above, discarding the unnecessary columns from any transformations since they have significant number of Nulls/NaNs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "c842d41f",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_with_no_spaces_in_colm_names = df_with_no_spaces_in_colm_names\\\n",
    ".drop('Days_Parking_In_Effect____'\\\n",
    "      ,'From_Hours_In_Effect'\\\n",
    "      ,'To_Hours_In_Effect'\\\n",
    "      ,'Unregistered_Vehicle?'\\\n",
    "      ,'No_Standing_or_Stopping_Violation'\\\n",
    "      ,'Hydrant_Violation'\\\n",
    "      ,'Double_Parking_Violation'\n",
    "     )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "e762e416",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Summons_Number',\n",
       " 'Plate_ID',\n",
       " 'Registration_State',\n",
       " 'Plate_Type',\n",
       " 'Violation_Code',\n",
       " 'Vehicle_Body_Type',\n",
       " 'Vehicle_Make',\n",
       " 'Issuing_Agency',\n",
       " 'Street_Code1',\n",
       " 'Street_Code2',\n",
       " 'Street_Code3',\n",
       " 'Vehicle_Expiration_Date',\n",
       " 'Violation_Location',\n",
       " 'Violation_Precinct',\n",
       " 'Issuer_Precinct',\n",
       " 'Issuer_Code',\n",
       " 'Issuer_Command',\n",
       " 'Issuer_Squad',\n",
       " 'Violation_Time',\n",
       " 'Time_First_Observed',\n",
       " 'Violation_County',\n",
       " 'Violation_In_Front_Of_Or_Opposite',\n",
       " 'House_Number',\n",
       " 'Street_Name',\n",
       " 'Intersecting_Street',\n",
       " 'Date_First_Observed',\n",
       " 'Law_Section',\n",
       " 'Sub_Division',\n",
       " 'Violation_Legal_Code',\n",
       " 'Vehicle_Color',\n",
       " 'Vehicle_Year',\n",
       " 'Meter_Number',\n",
       " 'Feet_From_Curb',\n",
       " 'Violation_Post_Code',\n",
       " 'Violation_Description',\n",
       " 'Violation_AM_or_PM',\n",
       " 'Issue_Date',\n",
       " 'Issue_Year',\n",
       " 'Issue_Month',\n",
       " 'Issue_DayofWeek',\n",
       " 'Issue_DayofMonth']"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6c67c7ac",
   "metadata": {},
   "source": [
    "#### Let's check the range of Issue Dates to ensure that the 2017 file don't have extra data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "d1605db9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 26:>                                                       (0 + 16) / 16]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------------+\n",
      "|min_Issue_Date|max_Issue_Date|\n",
      "+--------------+--------------+\n",
      "|    1972-03-30|    2069-11-19|\n",
      "+--------------+--------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names.\\\n",
    "agg(func.min('Issue_Date').alias('min_Issue_Date'), func.max('Issue_Date').alias('max_Issue_Date')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e042ebf",
   "metadata": {},
   "source": [
    "#### Trim the data down to 2017 year only"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "72b74238",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_2017_trimmed = df_with_no_spaces_in_colm_names.filter(df_with_no_spaces_in_colm_names.Issue_Year == '2017')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "70df64b0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Summons_Number: string, Plate_ID: string, Registration_State: string, Plate_Type: string, Violation_Code: string, Vehicle_Body_Type: string, Vehicle_Make: string, Issuing_Agency: string, Street_Code1: string, Street_Code2: string, Street_Code3: string, Vehicle_Expiration_Date: string, Violation_Location: string, Violation_Precinct: string, Issuer_Precinct: string, Issuer_Code: string, Issuer_Command: string, Issuer_Squad: string, Violation_Time: string, Time_First_Observed: string, Violation_County: string, Violation_In_Front_Of_Or_Opposite: string, House_Number: string, Street_Name: string, Intersecting_Street: string, Date_First_Observed: string, Law_Section: string, Sub_Division: string, Violation_Legal_Code: string, Vehicle_Color: string, Vehicle_Year: string, Meter_Number: string, Feet_From_Curb: string, Violation_Post_Code: string, Violation_Description: string, Violation_AM_or_PM: string, Issue_Date: string, Issue_Year: int, Issue_Month: int, Issue_DayofWeek: int, Issue_DayofMonth: int]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_with_no_spaces_in_colm_names.unpersist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "6a8f3599",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "5431918"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_2017_trimmed.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3559db0f",
   "metadata": {},
   "source": [
    "#### Let us create another categorical variable related to the hours of the day/night when violations have occured"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "1d1954de",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_2017_trimmed = df_2017_trimmed\\\n",
    ".withColumn('Violation_Time_Hour',func.substring(df_2017_trimmed.Violation_Time,1,2).cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "3b80ab20",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-------------------+------------------+\n",
      "|Violation_Time|Violation_Time_Hour|Violation_AM_or_PM|\n",
      "+--------------+-------------------+------------------+\n",
      "|         1120A|                 11|                 A|\n",
      "|         0852P|                  8|                 P|\n",
      "|         0015A|                  0|                 A|\n",
      "|         0525A|                  5|                 A|\n",
      "|         0256P|                  2|                 P|\n",
      "|         1232A|                 12|                 A|\n",
      "|         1034A|                 10|                 A|\n",
      "|         1021A|                 10|                 A|\n",
      "|         0721A|                  7|                 A|\n",
      "|         0940A|                  9|                 A|\n",
      "+--------------+-------------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_2017_trimmed.select('Violation_Time','Violation_Time_Hour','Violation_AM_or_PM').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "e93725ad",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_2017_trimmed.write.csv(path=clean_data_path, mode='overwrite', header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2ac87327",
   "metadata": {},
   "source": [
    "#### After the above preprocessing steps on 2017 data, we are ready to build necessary transformations/features on the processed/cleansed data. These transformations/features will help us answer the questions asked from this dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "bfbd4397",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test = spark.read.csv(clean_data_path, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "ae4604d8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------+------------------+----------+--------------+-----------------+------------+--------------+------------+------------+------------+-----------------------+------------------+------------------+---------------+-----------+--------------+------------+--------------+-------------------+----------------+---------------------------------+------------+--------------------+-------------------+-------------------+-----------+------------+--------------------+-------------+------------+------------+--------------+-------------------+---------------------+------------------+----------+----------+-----------+---------------+----------------+-------------------+\n",
      "|Summons_Number|Plate_ID|Registration_State|Plate_Type|Violation_Code|Vehicle_Body_Type|Vehicle_Make|Issuing_Agency|Street_Code1|Street_Code2|Street_Code3|Vehicle_Expiration_Date|Violation_Location|Violation_Precinct|Issuer_Precinct|Issuer_Code|Issuer_Command|Issuer_Squad|Violation_Time|Time_First_Observed|Violation_County|Violation_In_Front_Of_Or_Opposite|House_Number|         Street_Name|Intersecting_Street|Date_First_Observed|Law_Section|Sub_Division|Violation_Legal_Code|Vehicle_Color|Vehicle_Year|Meter_Number|Feet_From_Curb|Violation_Post_Code|Violation_Description|Violation_AM_or_PM|Issue_Date|Issue_Year|Issue_Month|Issue_DayofWeek|Issue_DayofMonth|Violation_Time_Hour|\n",
      "+--------------+--------+------------------+----------+--------------+-----------------+------------+--------------+------------+------------+------------+-----------------------+------------------+------------------+---------------+-----------+--------------+------------+--------------+-------------------+----------------+---------------------------------+------------+--------------------+-------------------+-------------------+-----------+------------+--------------------+-------------+------------+------------+--------------+-------------------+---------------------+------------------+----------+----------+-----------+---------------+----------------+-------------------+\n",
      "|    8524719369|  T236DP|                NY|       PAS|            71|             SUBN|       ROVER|             T|       61220|       29820|       66860|               20170904|              0049|                49|             49|     358145|          T201|           O|         1137A|               null|              BX|                                F|        1579|     Rhinelander Ave|               null|                  0|        408|          j6|                null|           GR|        1997|        null|             0|                  C| 71A-Insp Sticker ...|                 A|2017-05-01|      2017|          5|              2|               1|                 11|\n",
      "|    8527953651| GTV3226|                NY|       PAS|            38|             SUBN|         BMW|             T|       77150|       50020|       78485|               20180929|              0043|                43|             43|     363967|          T201|           F|         0522P|               null|              BX|                                F|        1354|     White Plains Rd|               null|                  0|        408|          h1|                null|           BK|        2015|        null|             0|                 03| 38-Failure to Dis...|                 P|2017-03-20|      2017|          3|              2|              20|                  5|\n",
      "|    4634047925| AEV8820|                NY|       PAS|            36|             SUBN|       LEXUS|             V|           0|           0|           0|                      0|              null|                 0|              0|          0|          null|        null|         1122A|               null|              BK|                             null|        null|WB FLATLANDS AVE @ E|             100 ST|                  0|       1180|           B|                   T|           WH|        2015|        null|             0|               null| PHTO SCHOOL ZN SP...|                 A|2017-05-18|      2017|          5|              5|              18|                 11|\n",
      "+--------------+--------+------------------+----------+--------------+-----------------+------------+--------------+------------+------------+------------+-----------------------+------------------+------------------+---------------+-----------+--------------+------------+--------------+-------------------+----------------+---------------------------------+------------+--------------------+-------------------+-------------------+-----------+------------+--------------------+-------------+------------+------------+--------------+-------------------+---------------------+------------------+----------+----------+-----------+---------------+----------------+-------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_test.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "093d1a55",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}