Capstone Project Template.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Project Title\n",
    "### Data Engineering Capstone Project\n",
    "\n",
    "#### Project Summary\n",
    "--describe your project at a high level--\n",
    "\n",
    "The project follows the follow steps:\n",
    "* Step 1: Scope the Project and Gather Data\n",
    "* Step 2: Explore and Assess the Data\n",
    "* Step 3: Define the Data Model\n",
    "* Step 4: Run ETL to Model the Data\n",
    "* Step 5: Complete Project Write Up"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Do all imports and installs here\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import configparser\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col,isnan,when,count,avg\n",
    "from pyspark.sql.types import TimestampType\n",
    "from pyspark.sql import functions as F\n",
    "import datetime\n",
    "\n",
    "def sas_date_to_datetime(sas_date):\n",
    "    '''\n",
    "    Converts given SAS numeric date to datetime\n",
    "    '''\n",
    "    if sas_date is None:\n",
    "        return None\n",
    "    return str(datetime.date(1960, 1, 1) + datetime.timedelta(days=sas_date))\n",
    "\n",
    "spark = SparkSession.builder.\\\n",
    "config(\"spark.jars.repositories\", \"https://repos.spark-packages.org/\").\\\n",
    "config(\"spark.jars.packages\", \"saurfang:spark-sas7bdat:2.0.0-s_2.11\").\\\n",
    "enableHiveSupport().getOrCreate()\n",
    "\n",
    "def generate_time_df(immigration_data):\n",
    "    '''\n",
    "    Returns a df with (timestamp,year,month,day,week,weekday) columns\n",
    "    From given immigrations df with (SAS date column(s))\n",
    "    '''\n",
    "    # get date columns\n",
    "    time_df = immigration_data.select(['arrival_date', 'departure_date'])\n",
    "\n",
    "    # Start creating unified date time df\n",
    "    arrival_df = time_df.select('arrival_date').dropDuplicates()\n",
    "    departure_df = time_df.select('departure_date').dropDuplicates()\n",
    "    unified_df = arrival_df.union(departure_df).dropDuplicates()\n",
    "\n",
    "    reg_convert_sas_date = F.udf(lambda date: sas_date_to_datetime(date))\n",
    "\n",
    "    # Apply sas date conversion function\n",
    "    unified_df = unified_df.withColumn('arrivalDateAsDATE', reg_convert_sas_date(unified_df.arrival_date))\n",
    "\n",
    "    # Add other columns\n",
    "    unified_df = unified_df.withColumn('year', F.year('arrivalDateAsDATE'))\n",
    "    unified_df = unified_df.withColumn('month', F.month('arrivalDateAsDATE'))\n",
    "    unified_df = unified_df.withColumn('day', F.dayofmonth('arrivalDateAsDATE'))\n",
    "    unified_df = unified_df.withColumn('week', F.weekofyear('arrivalDateAsDATE'))\n",
    "    unified_df = unified_df.withColumn('weekday', F.dayofweek('arrivalDateAsDATE'))\n",
    "\n",
    "    # Drop date string column since we no longer need it\n",
    "    unified_df = unified_df.drop('arrivalDateAsDATE')\n",
    "\n",
    "    # Rename \n",
    "    unified_df = unified_df.withColumnRenamed('arrival_date','timestamp')\n",
    "    unified_df = unified_df.withColumn('timestamp',unified_df['timestamp'].cast('int'))\n",
    "    \n",
    "    return unified_df\n",
    "\n",
    "# For immigration data\n",
    "# These columns are either meaningless for the scope of the project or almost empty, so we are removing them\n",
    "drop_cols = (\"i94yr\",\"i94mon\",\"i94res\",\"count\",\"visapost\",\"occup\",\"entdepa\",\"entdepd\",\"entdepu\",\"matflag\" \\\n",
    "             ,\"biryear\",\"insnum\",\"fltno\",\"dtadfile\",\"dtaddto\",\"airline\",\"admnum\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 1: Scope the Project and Gather Data\n",
    "\n",
    "#### Scope \n",
    "Explain what you plan to do in the project in more detail. What data do you use? What is your end solution look like? What tools did you use? etc>\n",
    "\n",
    "## Purpose\n",
    "This project aims to create a single source of truth database regarding I-94 information gathered throughout the year 2016.\n",
    "\n",
    "### Considerations\n",
    "* Unnamed column in immigration data will be dropped since there is no information about it\n",
    "\n",
    "#### Describe and Gather Data \n",
    "Describe the data sets you're using. Where did it come from? What type of information is included? "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# 1) Clean and explore cities, ports and immigrations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## 1.1) Read Cities"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- Median Age: double (nullable = true)\n",
      " |-- Male Population: integer (nullable = true)\n",
      " |-- Female Population: integer (nullable = true)\n",
      " |-- Total Population: integer (nullable = true)\n",
      " |-- Number of Veterans: integer (nullable = true)\n",
      " |-- Foreign-born: integer (nullable = true)\n",
      " |-- Average Household Size: double (nullable = true)\n",
      " |-- State Code: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>City</th>\n",
       "      <th>State</th>\n",
       "      <th>Median Age</th>\n",
       "      <th>Male Population</th>\n",
       "      <th>Female Population</th>\n",
       "      <th>Total Population</th>\n",
       "      <th>Number of Veterans</th>\n",
       "      <th>Foreign-born</th>\n",
       "      <th>Average Household Size</th>\n",
       "      <th>State Code</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Silver Spring</td>\n",
       "      <td>Maryland</td>\n",
       "      <td>33.8</td>\n",
       "      <td>40601</td>\n",
       "      <td>41862</td>\n",
       "      <td>82463</td>\n",
       "      <td>1562</td>\n",
       "      <td>30908</td>\n",
       "      <td>2.60</td>\n",
       "      <td>MD</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Quincy</td>\n",
       "      <td>Massachusetts</td>\n",
       "      <td>41.0</td>\n",
       "      <td>44129</td>\n",
       "      <td>49500</td>\n",
       "      <td>93629</td>\n",
       "      <td>4147</td>\n",
       "      <td>32935</td>\n",
       "      <td>2.39</td>\n",
       "      <td>MA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Hoover</td>\n",
       "      <td>Alabama</td>\n",
       "      <td>38.5</td>\n",
       "      <td>38040</td>\n",
       "      <td>46799</td>\n",
       "      <td>84839</td>\n",
       "      <td>4819</td>\n",
       "      <td>8229</td>\n",
       "      <td>2.58</td>\n",
       "      <td>AL</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Rancho Cucamonga</td>\n",
       "      <td>California</td>\n",
       "      <td>34.5</td>\n",
       "      <td>88127</td>\n",
       "      <td>87105</td>\n",
       "      <td>175232</td>\n",
       "      <td>5821</td>\n",
       "      <td>33878</td>\n",
       "      <td>3.18</td>\n",
       "      <td>CA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Newark</td>\n",
       "      <td>New Jersey</td>\n",
       "      <td>34.6</td>\n",
       "      <td>138040</td>\n",
       "      <td>143873</td>\n",
       "      <td>281913</td>\n",
       "      <td>5829</td>\n",
       "      <td>86253</td>\n",
       "      <td>2.73</td>\n",
       "      <td>NJ</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>Peoria</td>\n",
       "      <td>Illinois</td>\n",
       "      <td>33.1</td>\n",
       "      <td>56229</td>\n",
       "      <td>62432</td>\n",
       "      <td>118661</td>\n",
       "      <td>6634</td>\n",
       "      <td>7517</td>\n",
       "      <td>2.40</td>\n",
       "      <td>IL</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>Avondale</td>\n",
       "      <td>Arizona</td>\n",
       "      <td>29.1</td>\n",
       "      <td>38712</td>\n",
       "      <td>41971</td>\n",
       "      <td>80683</td>\n",
       "      <td>4815</td>\n",
       "      <td>8355</td>\n",
       "      <td>3.18</td>\n",
       "      <td>AZ</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>West Covina</td>\n",
       "      <td>California</td>\n",
       "      <td>39.8</td>\n",
       "      <td>51629</td>\n",
       "      <td>56860</td>\n",
       "      <td>108489</td>\n",
       "      <td>3800</td>\n",
       "      <td>37038</td>\n",
       "      <td>3.56</td>\n",
       "      <td>CA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>O'Fallon</td>\n",
       "      <td>Missouri</td>\n",
       "      <td>36.0</td>\n",
       "      <td>41762</td>\n",
       "      <td>43270</td>\n",
       "      <td>85032</td>\n",
       "      <td>5783</td>\n",
       "      <td>3269</td>\n",
       "      <td>2.77</td>\n",
       "      <td>MO</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>High Point</td>\n",
       "      <td>North Carolina</td>\n",
       "      <td>35.5</td>\n",
       "      <td>51751</td>\n",
       "      <td>58077</td>\n",
       "      <td>109828</td>\n",
       "      <td>5204</td>\n",
       "      <td>16315</td>\n",
       "      <td>2.65</td>\n",
       "      <td>NC</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "               City           State  Median Age  Male Population  \\\n",
       "0     Silver Spring        Maryland        33.8            40601   \n",
       "1            Quincy   Massachusetts        41.0            44129   \n",
       "2            Hoover         Alabama        38.5            38040   \n",
       "3  Rancho Cucamonga      California        34.5            88127   \n",
       "4            Newark      New Jersey        34.6           138040   \n",
       "5            Peoria        Illinois        33.1            56229   \n",
       "6          Avondale         Arizona        29.1            38712   \n",
       "7       West Covina      California        39.8            51629   \n",
       "8          O'Fallon        Missouri        36.0            41762   \n",
       "9        High Point  North Carolina        35.5            51751   \n",
       "\n",
       "   Female Population  Total Population  Number of Veterans  Foreign-born  \\\n",
       "0              41862             82463                1562         30908   \n",
       "1              49500             93629                4147         32935   \n",
       "2              46799             84839                4819          8229   \n",
       "3              87105            175232                5821         33878   \n",
       "4             143873            281913                5829         86253   \n",
       "5              62432            118661                6634          7517   \n",
       "6              41971             80683                4815          8355   \n",
       "7              56860            108489                3800         37038   \n",
       "8              43270             85032                5783          3269   \n",
       "9              58077            109828                5204         16315   \n",
       "\n",
       "   Average Household Size State Code  \n",
       "0                    2.60         MD  \n",
       "1                    2.39         MA  \n",
       "2                    2.58         AL  \n",
       "3                    3.18         CA  \n",
       "4                    2.73         NJ  \n",
       "5                    2.40         IL  \n",
       "6                    3.18         AZ  \n",
       "7                    3.56         CA  \n",
       "8                    2.77         MO  \n",
       "9                    2.65         NC  "
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Explore us-cities-demographics.csv\n",
    "# read data into frame read csv\n",
    "city_df = spark.read.options(header=\"true\",inferSchema=\"true\",nullValue = \"NULL\",delimiter=\";\").csv('us-cities-demographics.csv')\n",
    "\n",
    "# drop race and count columns\n",
    "city_df = city_df.drop('Race','Count')\n",
    "\n",
    "# Get non-null state code and city\n",
    "city_df = city_df.filter(city_df['State Code'].isNotNull() & city_df['City'].isNotNull())\n",
    "\n",
    "city_df.printSchema()\n",
    "city_df.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## 1.2) Read Ports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ident: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- continent: string (nullable = true)\n",
      " |-- iso_region: string (nullable = true)\n",
      " |-- municipality: string (nullable = true)\n",
      " |-- iata_code: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>ident</th>\n",
       "      <th>type</th>\n",
       "      <th>name</th>\n",
       "      <th>continent</th>\n",
       "      <th>iso_region</th>\n",
       "      <th>municipality</th>\n",
       "      <th>iata_code</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>07FA</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Ocean Reef Club Airport</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-FL</td>\n",
       "      <td>Key Largo</td>\n",
       "      <td>OCA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0AK</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Pilot Station Airport</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Pilot Station</td>\n",
       "      <td>PQS</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>0CO2</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Crested Butte Airpark</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-CO</td>\n",
       "      <td>Crested Butte</td>\n",
       "      <td>CSE</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>0TE7</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>LBJ Ranch Airport</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-TX</td>\n",
       "      <td>Johnson City</td>\n",
       "      <td>JCY</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>13MA</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Metropolitan Airport</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-MA</td>\n",
       "      <td>Palmer</td>\n",
       "      <td>PMX</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>13Z</td>\n",
       "      <td>seaplane_base</td>\n",
       "      <td>Loring Seaplane Base</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Loring</td>\n",
       "      <td>WLR</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>16A</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Nunapitchuk Airport</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Nunapitchuk</td>\n",
       "      <td>NUP</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>16K</td>\n",
       "      <td>seaplane_base</td>\n",
       "      <td>Port Alice Seaplane Base</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Port Alice</td>\n",
       "      <td>PTC</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>19AK</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Icy Bay Airport</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Icy Bay</td>\n",
       "      <td>ICY</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>19P</td>\n",
       "      <td>seaplane_base</td>\n",
       "      <td>Port Protection Seaplane Base</td>\n",
       "      <td>NA</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Port Protection</td>\n",
       "      <td>PPV</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  ident           type                           name continent iso_region  \\\n",
       "0  07FA  small_airport        Ocean Reef Club Airport        NA      US-FL   \n",
       "1   0AK  small_airport          Pilot Station Airport        NA      US-AK   \n",
       "2  0CO2  small_airport          Crested Butte Airpark        NA      US-CO   \n",
       "3  0TE7  small_airport              LBJ Ranch Airport        NA      US-TX   \n",
       "4  13MA  small_airport           Metropolitan Airport        NA      US-MA   \n",
       "5   13Z  seaplane_base           Loring Seaplane Base        NA      US-AK   \n",
       "6   16A  small_airport            Nunapitchuk Airport        NA      US-AK   \n",
       "7   16K  seaplane_base       Port Alice Seaplane Base        NA      US-AK   \n",
       "8  19AK  small_airport                Icy Bay Airport        NA      US-AK   \n",
       "9   19P  seaplane_base  Port Protection Seaplane Base        NA      US-AK   \n",
       "\n",
       "      municipality iata_code  \n",
       "0        Key Largo       OCA  \n",
       "1    Pilot Station       PQS  \n",
       "2    Crested Butte       CSE  \n",
       "3     Johnson City       JCY  \n",
       "4           Palmer       PMX  \n",
       "5           Loring       WLR  \n",
       "6      Nunapitchuk       NUP  \n",
       "7       Port Alice       PTC  \n",
       "8          Icy Bay       ICY  \n",
       "9  Port Protection       PPV  "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# read\n",
    "ports_df = spark.read.options(header=\"true\",inferSchema=\"true\",nullValue = \"NULL\").csv('airport-codes_csv.csv')\n",
    "\n",
    "# Record count is 55075 initally\n",
    "# Lets take iso_country US, non-null iata code and non-closed records\n",
    "ports_df = ports_df.filter((ports_df.iata_code.isNotNull()) \\\n",
    "                            & (ports_df.iso_country == 'US') \\\n",
    "                            & (ports_df.type != 'closed') )\n",
    "\n",
    "drop_cols = ('iso_country','gps_code','elevation_ft','local_code','coordinates')\n",
    "\n",
    "ports_df = ports_df.drop(*drop_cols)\n",
    "\n",
    "# Show some info\n",
    "ports_df.printSchema()\n",
    "ports_df.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## 1.3) Combine cities and ports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- Median Age: double (nullable = true)\n",
      " |-- Male Population: integer (nullable = true)\n",
      " |-- Female Population: integer (nullable = true)\n",
      " |-- Total Population: integer (nullable = true)\n",
      " |-- Number of Veterans: integer (nullable = true)\n",
      " |-- Foreign-born: integer (nullable = true)\n",
      " |-- Average Household Size: double (nullable = true)\n",
      " |-- State Code: string (nullable = true)\n",
      " |-- ident: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- continent: string (nullable = true)\n",
      " |-- iso_region: string (nullable = true)\n",
      " |-- municipality: string (nullable = true)\n",
      " |-- iata_code: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Join on 'City.city == ports_df.municipality\n",
    "combined_df = city_df.join(ports_df, city_df.City == ports_df.municipality).dropDuplicates()\n",
    "combined_df.printSchema() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Now cities have been matched!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## 1.4) Finalize ports and cities"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 1.4.1) Ports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- port_id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- iata_code: string (nullable = true)\n",
      " |-- iso_region: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>port_id</th>\n",
       "      <th>name</th>\n",
       "      <th>city</th>\n",
       "      <th>type</th>\n",
       "      <th>iata_code</th>\n",
       "      <th>iso_region</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>KBFI</td>\n",
       "      <td>Boeing Field King County International Airport</td>\n",
       "      <td>Seattle</td>\n",
       "      <td>large_airport</td>\n",
       "      <td>BFI</td>\n",
       "      <td>WA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>KVSF</td>\n",
       "      <td>Hartness State (Springfield) Airport</td>\n",
       "      <td>Springfield</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>VSF</td>\n",
       "      <td>VT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>KSAF</td>\n",
       "      <td>Santa Fe Municipal Airport</td>\n",
       "      <td>Santa Fe</td>\n",
       "      <td>medium_airport</td>\n",
       "      <td>SAF</td>\n",
       "      <td>NM</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>KCHA</td>\n",
       "      <td>Lovell Field</td>\n",
       "      <td>Chattanooga</td>\n",
       "      <td>large_airport</td>\n",
       "      <td>CHA</td>\n",
       "      <td>TN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>KMBO</td>\n",
       "      <td>Bruce Campbell Field</td>\n",
       "      <td>Madison</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>DXE</td>\n",
       "      <td>MS</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>KOFF</td>\n",
       "      <td>Offutt Air Force Base</td>\n",
       "      <td>Omaha</td>\n",
       "      <td>medium_airport</td>\n",
       "      <td>OFF</td>\n",
       "      <td>NE</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>KTTD</td>\n",
       "      <td>Portland Troutdale Airport</td>\n",
       "      <td>Portland</td>\n",
       "      <td>medium_airport</td>\n",
       "      <td>TTD</td>\n",
       "      <td>OR</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>KWHP</td>\n",
       "      <td>Whiteman Airport</td>\n",
       "      <td>Los Angeles</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>WHP</td>\n",
       "      <td>CA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>KLAF</td>\n",
       "      <td>Purdue University Airport</td>\n",
       "      <td>Lafayette</td>\n",
       "      <td>medium_airport</td>\n",
       "      <td>LAF</td>\n",
       "      <td>IN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>KSHV</td>\n",
       "      <td>Shreveport Regional Airport</td>\n",
       "      <td>Shreveport</td>\n",
       "      <td>medium_airport</td>\n",
       "      <td>SHV</td>\n",
       "      <td>LA</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  port_id                                            name         city  \\\n",
       "0    KBFI  Boeing Field King County International Airport      Seattle   \n",
       "1    KVSF            Hartness State (Springfield) Airport  Springfield   \n",
       "2    KSAF                      Santa Fe Municipal Airport     Santa Fe   \n",
       "3    KCHA                                    Lovell Field  Chattanooga   \n",
       "4    KMBO                            Bruce Campbell Field      Madison   \n",
       "5    KOFF                           Offutt Air Force Base        Omaha   \n",
       "6    KTTD                      Portland Troutdale Airport     Portland   \n",
       "7    KWHP                                Whiteman Airport  Los Angeles   \n",
       "8    KLAF                       Purdue University Airport    Lafayette   \n",
       "9    KSHV                     Shreveport Regional Airport   Shreveport   \n",
       "\n",
       "             type iata_code iso_region  \n",
       "0   large_airport       BFI         WA  \n",
       "1   small_airport       VSF         VT  \n",
       "2  medium_airport       SAF         NM  \n",
       "3   large_airport       CHA         TN  \n",
       "4   small_airport       DXE         MS  \n",
       "5  medium_airport       OFF         NE  \n",
       "6  medium_airport       TTD         OR  \n",
       "7   small_airport       WHP         CA  \n",
       "8  medium_airport       LAF         IN  \n",
       "9  medium_airport       SHV         LA  "
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Define a function to get state part from 'iso_region' column which is the second item of the list after split by '-'\n",
    "def split_iso_region(iso_region):\n",
    "    return iso_region.split(\"-\")[1]\n",
    "\n",
    "# Register split function\n",
    "reg_split_iso_region = F.udf(lambda iso_reg: split_iso_region(iso_reg))\n",
    "\n",
    "\n",
    "final_ports_df = combined_df.select(['ident','name','municipality','type','iata_code','iso_region']) \\\n",
    "                            .withColumnRenamed('ident','port_id') \\\n",
    "                            .withColumnRenamed('municipality','city') \n",
    "\n",
    "# Apply split function to iso_region\n",
    "final_ports_df = final_ports_df.withColumn('iso_region',reg_split_iso_region('iso_region'))\n",
    "final_ports_df.printSchema()\n",
    "final_ports_df.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 1.4.2) Cities"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- city: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- median_age: double (nullable = true)\n",
      " |-- male_pop: integer (nullable = true)\n",
      " |-- female_pop: integer (nullable = true)\n",
      " |-- total_pop: integer (nullable = true)\n",
      " |-- veterans: integer (nullable = true)\n",
      " |-- foreign_born: integer (nullable = true)\n",
      " |-- avg_household_size: double (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>city</th>\n",
       "      <th>state</th>\n",
       "      <th>median_age</th>\n",
       "      <th>male_pop</th>\n",
       "      <th>female_pop</th>\n",
       "      <th>total_pop</th>\n",
       "      <th>veterans</th>\n",
       "      <th>foreign_born</th>\n",
       "      <th>avg_household_size</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Seattle</td>\n",
       "      <td>Washington</td>\n",
       "      <td>35.5</td>\n",
       "      <td>345659</td>\n",
       "      <td>338784</td>\n",
       "      <td>684443</td>\n",
       "      <td>29364</td>\n",
       "      <td>119840</td>\n",
       "      <td>2.13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Springfield</td>\n",
       "      <td>Illinois</td>\n",
       "      <td>38.8</td>\n",
       "      <td>55639</td>\n",
       "      <td>62170</td>\n",
       "      <td>117809</td>\n",
       "      <td>7525</td>\n",
       "      <td>4264</td>\n",
       "      <td>2.22</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Santa Fe</td>\n",
       "      <td>New Mexico</td>\n",
       "      <td>44.1</td>\n",
       "      <td>40601</td>\n",
       "      <td>43511</td>\n",
       "      <td>84112</td>\n",
       "      <td>5083</td>\n",
       "      <td>13824</td>\n",
       "      <td>2.41</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Chattanooga</td>\n",
       "      <td>Tennessee</td>\n",
       "      <td>36.6</td>\n",
       "      <td>83640</td>\n",
       "      <td>92957</td>\n",
       "      <td>176597</td>\n",
       "      <td>10001</td>\n",
       "      <td>10599</td>\n",
       "      <td>2.40</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Madison</td>\n",
       "      <td>Wisconsin</td>\n",
       "      <td>30.7</td>\n",
       "      <td>122596</td>\n",
       "      <td>126360</td>\n",
       "      <td>248956</td>\n",
       "      <td>9707</td>\n",
       "      <td>30090</td>\n",
       "      <td>2.23</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>Omaha</td>\n",
       "      <td>Nebraska</td>\n",
       "      <td>34.2</td>\n",
       "      <td>218789</td>\n",
       "      <td>225098</td>\n",
       "      <td>443887</td>\n",
       "      <td>24503</td>\n",
       "      <td>48263</td>\n",
       "      <td>2.47</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>Portland</td>\n",
       "      <td>Oregon</td>\n",
       "      <td>36.7</td>\n",
       "      <td>313516</td>\n",
       "      <td>318671</td>\n",
       "      <td>632187</td>\n",
       "      <td>29940</td>\n",
       "      <td>86041</td>\n",
       "      <td>2.43</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>Los Angeles</td>\n",
       "      <td>California</td>\n",
       "      <td>35.0</td>\n",
       "      <td>1958998</td>\n",
       "      <td>2012898</td>\n",
       "      <td>3971896</td>\n",
       "      <td>85417</td>\n",
       "      <td>1485425</td>\n",
       "      <td>2.86</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>Lafayette</td>\n",
       "      <td>Indiana</td>\n",
       "      <td>33.5</td>\n",
       "      <td>34313</td>\n",
       "      <td>36857</td>\n",
       "      <td>71170</td>\n",
       "      <td>5045</td>\n",
       "      <td>5697</td>\n",
       "      <td>2.19</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>Shreveport</td>\n",
       "      <td>Louisiana</td>\n",
       "      <td>35.2</td>\n",
       "      <td>93138</td>\n",
       "      <td>103856</td>\n",
       "      <td>196994</td>\n",
       "      <td>14287</td>\n",
       "      <td>5658</td>\n",
       "      <td>2.53</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          city       state  median_age  male_pop  female_pop  total_pop  \\\n",
       "0      Seattle  Washington        35.5    345659      338784     684443   \n",
       "1  Springfield    Illinois        38.8     55639       62170     117809   \n",
       "2     Santa Fe  New Mexico        44.1     40601       43511      84112   \n",
       "3  Chattanooga   Tennessee        36.6     83640       92957     176597   \n",
       "4      Madison   Wisconsin        30.7    122596      126360     248956   \n",
       "5        Omaha    Nebraska        34.2    218789      225098     443887   \n",
       "6     Portland      Oregon        36.7    313516      318671     632187   \n",
       "7  Los Angeles  California        35.0   1958998     2012898    3971896   \n",
       "8    Lafayette     Indiana        33.5     34313       36857      71170   \n",
       "9   Shreveport   Louisiana        35.2     93138      103856     196994   \n",
       "\n",
       "   veterans  foreign_born  avg_household_size  \n",
       "0     29364        119840                2.13  \n",
       "1      7525          4264                2.22  \n",
       "2      5083         13824                2.41  \n",
       "3     10001         10599                2.40  \n",
       "4      9707         30090                2.23  \n",
       "5     24503         48263                2.47  \n",
       "6     29940         86041                2.43  \n",
       "7     85417       1485425                2.86  \n",
       "8      5045          5697                2.19  \n",
       "9     14287          5658                2.53  "
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "final_cities_df = combined_df.select(['City','State','Median Age','Male Population','Female Population','Total Population' \\\n",
    "                                      ,'Number of Veterans','Foreign-born','Average Household Size']) \\\n",
    "                            .withColumnRenamed('City','city') \\\n",
    "                            .withColumnRenamed('State','state') \\\n",
    "                            .withColumnRenamed('Median Age','median_age') \\\n",
    "                            .withColumnRenamed('Male Population','male_pop') \\\n",
    "                            .withColumnRenamed('Female Population','female_pop') \\\n",
    "                            .withColumnRenamed('Total Population','total_pop') \\\n",
    "                            .withColumnRenamed('Number of Veterans','veterans') \\\n",
    "                            .withColumnRenamed('Foreign-born','foreign_born') \\\n",
    "                            .withColumnRenamed('Average Household Size','avg_household_size')\n",
    "final_cities_df.printSchema()\n",
    "final_cities_df.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## 1.5) Immigrations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Process immigrations data one by one in a loop, each iteration:\n",
    "* Read data\n",
    "* Repartition\n",
    "* Drop duplicates, redundant columns, null records\n",
    "* Finally,\n",
    "    * JOIN ON conditions list = __[df.state == final_ports_df.iso_region, df.landing_port == final_ports_df.iata_code]__\n",
    "    * joined_immig = df.join(final_ports_df, conditions)\n",
    "    * Additionally, create time data using joined_immig (using arrdate and depdate columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# These columns are either meaningless for the scope of the project or almost empty, so we are removing them\n",
    "drop_cols = (\"i94yr\",\"i94mon\",\"i94res\",\"count\",\"visapost\",\"occup\",\"entdepa\",\"entdepd\",\"entdepu\",\"matflag\" \\\n",
    "             ,\"biryear\",\"insnum\",\"fltno\",\"dtadfile\",\"dtaddto\",\"airline\",\"admnum\")\n",
    "\n",
    "# Month name list\n",
    "months = ['jan','feb','mar','apr','may','jun','jul','aug','sep','oct','nov','dec']\n",
    "\n",
    "#for month in months:\n",
    "#immigration_data_path = f'../../data/18-83510-I94-Data-2016/i94_{month}16_sub.sas7bdat'\n",
    "immigration_data_path = f'../../data/18-83510-I94-Data-2016/i94_feb16_sub.sas7bdat'\n",
    "\n",
    "# Read data using month in input file name\n",
    "df = spark.read.format('com.github.saurfang.sas.spark').load(immigration_data_path)\n",
    "df.repartition(6)\n",
    "# Drop duplicates, redundant columns, null records from each df and write to S3 in parquet format\n",
    "df = df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "\n",
    "# Fix column names and types\n",
    "df = df.withColumnRenamed('cicid','immigration_id')\n",
    "df = df.withColumn('immigration_id',df['immigration_id'].cast('int'))\n",
    "df = df.withColumnRenamed('i94cit','origin')\n",
    "df = df.withColumn('origin',df['origin'].cast('int'))\n",
    "df = df.withColumnRenamed('i94bir','age')\n",
    "df = df.withColumn('age',df['age'].cast('int'))\n",
    "df = df.withColumnRenamed('i94mode','arrival_mode')\n",
    "df = df.withColumn('arrival_mode',df['arrival_mode'].cast('int'))\n",
    "df = df.withColumnRenamed('i94visa','visa')\n",
    "df = df.withColumn('visa',df['visa'].cast('int'))\n",
    "df = df.withColumnRenamed('i94port','landing_port') \\\n",
    "    .withColumnRenamed('arrdate','arrival_date') \\\n",
    "    .withColumnRenamed('i94mode','arrival_mode') \\\n",
    "    .withColumnRenamed('i94addr','state') \\\n",
    "    .withColumnRenamed('depdate','departure_date')\n",
    "\n",
    "# Now generate time df from current immigration data using 'arrival_date' and 'departure_date' columns\n",
    "#time_df = generate_time_df(df)\n",
    "#time_df = time_df.na.drop()\n",
    "#time_df.printSchema()\n",
    "#time_df.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Join conditions as a list\n",
    "conditions = [df.state == final_ports_df.iso_region, df.landing_port == final_ports_df.iata_code]\n",
    "\n",
    "joined_immig = df.join(final_ports_df, conditions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- immigration_id: integer (nullable = true)\n",
      " |-- origin: integer (nullable = true)\n",
      " |-- landing_port: string (nullable = true)\n",
      " |-- arrival_date: double (nullable = true)\n",
      " |-- departure_date: double (nullable = true)\n",
      " |-- arrival_mode: integer (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      " |-- visa: integer (nullable = true)\n",
      " |-- visatype: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>immigration_id</th>\n",
       "      <th>origin</th>\n",
       "      <th>landing_port</th>\n",
       "      <th>arrival_date</th>\n",
       "      <th>departure_date</th>\n",
       "      <th>arrival_mode</th>\n",
       "      <th>city</th>\n",
       "      <th>age</th>\n",
       "      <th>visa</th>\n",
       "      <th>visatype</th>\n",
       "      <th>gender</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>75354</td>\n",
       "      <td>107</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20485.0</td>\n",
       "      <td>20488.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>28</td>\n",
       "      <td>2</td>\n",
       "      <td>B2</td>\n",
       "      <td>F</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>81940</td>\n",
       "      <td>116</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20485.0</td>\n",
       "      <td>20486.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>40</td>\n",
       "      <td>1</td>\n",
       "      <td>WB</td>\n",
       "      <td>M</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>85982</td>\n",
       "      <td>123</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20485.0</td>\n",
       "      <td>20492.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>64</td>\n",
       "      <td>2</td>\n",
       "      <td>WT</td>\n",
       "      <td>F</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>126885</td>\n",
       "      <td>245</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20485.0</td>\n",
       "      <td>20491.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>35</td>\n",
       "      <td>2</td>\n",
       "      <td>B2</td>\n",
       "      <td>F</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>294117</td>\n",
       "      <td>213</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20486.0</td>\n",
       "      <td>20488.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>59</td>\n",
       "      <td>1</td>\n",
       "      <td>B1</td>\n",
       "      <td>M</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>304039</td>\n",
       "      <td>245</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20486.0</td>\n",
       "      <td>20490.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>26</td>\n",
       "      <td>2</td>\n",
       "      <td>B2</td>\n",
       "      <td>F</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>304212</td>\n",
       "      <td>245</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20486.0</td>\n",
       "      <td>20599.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>26</td>\n",
       "      <td>3</td>\n",
       "      <td>F1</td>\n",
       "      <td>F</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>307803</td>\n",
       "      <td>251</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20486.0</td>\n",
       "      <td>20500.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>66</td>\n",
       "      <td>2</td>\n",
       "      <td>B2</td>\n",
       "      <td>M</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>435173</td>\n",
       "      <td>148</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20487.0</td>\n",
       "      <td>20495.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>24</td>\n",
       "      <td>2</td>\n",
       "      <td>WT</td>\n",
       "      <td>F</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>460876</td>\n",
       "      <td>245</td>\n",
       "      <td>BOS</td>\n",
       "      <td>20487.0</td>\n",
       "      <td>20505.0</td>\n",
       "      <td>1</td>\n",
       "      <td>Boston</td>\n",
       "      <td>54</td>\n",
       "      <td>2</td>\n",
       "      <td>B2</td>\n",
       "      <td>M</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   immigration_id  origin landing_port  arrival_date  departure_date  \\\n",
       "0           75354     107          BOS       20485.0         20488.0   \n",
       "1           81940     116          BOS       20485.0         20486.0   \n",
       "2           85982     123          BOS       20485.0         20492.0   \n",
       "3          126885     245          BOS       20485.0         20491.0   \n",
       "4          294117     213          BOS       20486.0         20488.0   \n",
       "5          304039     245          BOS       20486.0         20490.0   \n",
       "6          304212     245          BOS       20486.0         20599.0   \n",
       "7          307803     251          BOS       20486.0         20500.0   \n",
       "8          435173     148          BOS       20487.0         20495.0   \n",
       "9          460876     245          BOS       20487.0         20505.0   \n",
       "\n",
       "   arrival_mode    city  age  visa visatype gender  \n",
       "0             1  Boston   28     2       B2      F  \n",
       "1             1  Boston   40     1       WB      M  \n",
       "2             1  Boston   64     2       WT      F  \n",
       "3             1  Boston   35     2       B2      F  \n",
       "4             1  Boston   59     1       B1      M  \n",
       "5             1  Boston   26     2       B2      F  \n",
       "6             1  Boston   26     3       F1      F  \n",
       "7             1  Boston   66     2       B2      M  \n",
       "8             1  Boston   24     2       WT      F  \n",
       "9             1  Boston   54     2       B2      M  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Final df to write to S3\n",
    "fact_immig_df = joined_immig.select('immigration_id','origin','landing_port' \\\n",
    "                                    ,'arrival_date','departure_date','arrival_mode' \\\n",
    "                                    ,'city','age','visa','visatype','gender')\n",
    "fact_immig_df.printSchema()\n",
    "fact_immig_df.limit(10).toPandas()\n",
    "#print(f'Count: {fact_immig_df.count()}') # yields 466855"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "editable": true
   },
   "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>arrival_date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>20503.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>20511.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>20593.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>20522.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>20550.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>20592.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>20496.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>20518.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>20556.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>20485.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>20486.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>20553.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>20527.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>20584.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>14</th>\n",
       "      <td>20594.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>15</th>\n",
       "      <td>20497.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>16</th>\n",
       "      <td>20551.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>17</th>\n",
       "      <td>20565.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>18</th>\n",
       "      <td>20523.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>19</th>\n",
       "      <td>20544.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    arrival_date\n",
       "0        20503.0\n",
       "1        20511.0\n",
       "2        20593.0\n",
       "3        20522.0\n",
       "4        20550.0\n",
       "5        20592.0\n",
       "6        20496.0\n",
       "7        20518.0\n",
       "8        20556.0\n",
       "9        20485.0\n",
       "10       20486.0\n",
       "11       20553.0\n",
       "12       20527.0\n",
       "13       20584.0\n",
       "14       20594.0\n",
       "15       20497.0\n",
       "16       20551.0\n",
       "17       20565.0\n",
       "18       20523.0\n",
       "19       20544.0"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    " # get date columns\n",
    "time_df = fact_immig_df.select(['arrival_date', 'departure_date'])\n",
    "\n",
    "# Start creating unified date time df\n",
    "arrival_df = time_df.select('arrival_date').dropDuplicates()\n",
    "departure_df = time_df.select('departure_date').dropDuplicates()\n",
    "unified_df = arrival_df.union(departure_df).dropDuplicates()\n",
    "\n",
    "reg_convert_sas_date = F.udf(lambda date: sas_date_to_datetime(date))\n",
    "\n",
    "unified_df.limit(20).toPandas()\n",
    "\n",
    "# Apply sas date conversion function\n",
    "unified_df = unified_df.withColumn('arrivalDateAsDATE', reg_convert_sas_date(unified_df.arrival_date))\n",
    "\n",
    "# Add other columns\n",
    "#unified_df = unified_df.withColumn('year', F.year('arrivalDateAsDATE'))\n",
    "#unified_df = unified_df.withColumn('month', F.month('arrivalDateAsDATE'))\n",
    "#unified_df = unified_df.withColumn('day', F.dayofmonth('arrivalDateAsDATE'))\n",
    "#unified_df = unified_df.withColumn('week', F.weekofyear('arrivalDateAsDATE'))\n",
    "#unified_df = unified_df.withColumn('weekday', F.dayofweek('arrivalDateAsDATE'))\n",
    "\n",
    "# Drop date string column since we no longer need it\n",
    "#unified_df = unified_df.drop('arrivalDateAsDATE')\n",
    "\n",
    "# Rename \n",
    "#unified_df = unified_df.withColumnRenamed('arrival_date','sas_timestamp')\n",
    "\n",
    "#unified_df.show(50)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "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>arrival_date</th>\n",
       "      <th>arrivalDateAsDATE</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>20503.0</td>\n",
       "      <td>2016-02-19</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>20511.0</td>\n",
       "      <td>2016-02-27</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>20593.0</td>\n",
       "      <td>2016-05-19</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>20522.0</td>\n",
       "      <td>2016-03-09</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>20550.0</td>\n",
       "      <td>2016-04-06</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>20592.0</td>\n",
       "      <td>2016-05-18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>20496.0</td>\n",
       "      <td>2016-02-12</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>20518.0</td>\n",
       "      <td>2016-03-05</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>20556.0</td>\n",
       "      <td>2016-04-12</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>20485.0</td>\n",
       "      <td>2016-02-01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>20486.0</td>\n",
       "      <td>2016-02-02</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>20553.0</td>\n",
       "      <td>2016-04-09</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>20527.0</td>\n",
       "      <td>2016-03-14</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>20584.0</td>\n",
       "      <td>2016-05-10</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>14</th>\n",
       "      <td>20594.0</td>\n",
       "      <td>2016-05-20</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>15</th>\n",
       "      <td>20497.0</td>\n",
       "      <td>2016-02-13</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>16</th>\n",
       "      <td>20551.0</td>\n",
       "      <td>2016-04-07</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>17</th>\n",
       "      <td>20565.0</td>\n",
       "      <td>2016-04-21</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>18</th>\n",
       "      <td>20523.0</td>\n",
       "      <td>2016-03-10</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>19</th>\n",
       "      <td>20544.0</td>\n",
       "      <td>2016-03-31</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    arrival_date arrivalDateAsDATE\n",
       "0        20503.0        2016-02-19\n",
       "1        20511.0        2016-02-27\n",
       "2        20593.0        2016-05-19\n",
       "3        20522.0        2016-03-09\n",
       "4        20550.0        2016-04-06\n",
       "5        20592.0        2016-05-18\n",
       "6        20496.0        2016-02-12\n",
       "7        20518.0        2016-03-05\n",
       "8        20556.0        2016-04-12\n",
       "9        20485.0        2016-02-01\n",
       "10       20486.0        2016-02-02\n",
       "11       20553.0        2016-04-09\n",
       "12       20527.0        2016-03-14\n",
       "13       20584.0        2016-05-10\n",
       "14       20594.0        2016-05-20\n",
       "15       20497.0        2016-02-13\n",
       "16       20551.0        2016-04-07\n",
       "17       20565.0        2016-04-21\n",
       "18       20523.0        2016-03-10\n",
       "19       20544.0        2016-03-31"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Apply sas date conversion function\n",
    "unified_df = unified_df.withColumn('arrivalDateAsDATE', reg_convert_sas_date(unified_df.arrival_date))\n",
    "unified_df.limit(20).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----+-----+---+----+-------+\n",
      "|timestamp|year|month|day|week|weekday|\n",
      "+---------+----+-----+---+----+-------+\n",
      "|        0|   0|    0|  0|   0|      0|\n",
      "+---------+----+-----+---+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "time_df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in time_df.columns]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "118"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "time_df.na.drop().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Skewness check for some suspicious columns!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------+\n",
      "|visa| count|\n",
      "+----+------+\n",
      "| 2.0|396818|\n",
      "| 1.0| 66865|\n",
      "| 3.0|  3172|\n",
      "+----+------+\n",
      "\n",
      "+--------+------+\n",
      "|visatype| count|\n",
      "+--------+------+\n",
      "|      B2|260941|\n",
      "|      WT|135794|\n",
      "|      B1| 36372|\n",
      "|      WB| 26666|\n",
      "|      E2|  2910|\n",
      "|      F1|  2626|\n",
      "|      E1|   513|\n",
      "|      F2|   401|\n",
      "|       I|   383|\n",
      "|      M1|   144|\n",
      "|      CP|    77|\n",
      "|      I1|    21|\n",
      "|     CPL|     6|\n",
      "|      M2|     1|\n",
      "+--------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fact_immig_df.groupBy('visa').count().orderBy('count', ascending=False).show()\n",
    "fact_immig_df.groupBy('visatype').count().orderBy('count', ascending=False).show()\n",
    "# i94visa seems skewed a bit, let's repartition\n",
    "#df.rdd.getNumPartitions()\n",
    "#partitioned_df = df.repartition('visa')\n",
    "#partitioned_df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+------+\n",
      "|arrival_mode| count|\n",
      "+------------+------+\n",
      "|         1.0|465131|\n",
      "|         2.0|  1014|\n",
      "|         3.0|   650|\n",
      "|         9.0|    60|\n",
      "+------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fact_immig_df.groupBy('arrival_mode').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "200"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fact_immig_df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "200"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fact_immig_df = fact_immig_df.repartition(col('visa'))\n",
    "fact_immig_df.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# 2) Clean and explore GlobalLandandTemperaturesByCity\n",
    "* Filter by 'United States'\n",
    "* Drop redundant colums and rename the remaining ones\n",
    "* Drop rows with null values\n",
    "* Calculate the average of temperature and temperature uncertainty and group by 'city'\n",
    "* Resulting data frame can be written to S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- city: string (nullable = true)\n",
      " |-- avg_temp: double (nullable = true)\n",
      " |-- avg_temp_uncertainty: double (nullable = true)\n",
      "\n",
      "+-----------+------------------+--------------------+\n",
      "|       city|          avg_temp|avg_temp_uncertainty|\n",
      "+-----------+------------------+--------------------+\n",
      "|  Worcester| 7.341440525809558|  1.3742648284706618|\n",
      "| Charleston|18.696557871112546|  1.4356107726835539|\n",
      "|     Corona| 16.12483712696008|  0.7674734446130481|\n",
      "|Springfield|10.647931343609901|  1.3296092707991722|\n",
      "|      Tempe|  21.0487690509584|  0.7654862085086479|\n",
      "+-----------+------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "248"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "temps_df = spark.read.options(header=\"true\",inferSchema=\"true\",nullValue = \"NULL\").csv('../../data2/GlobalLandTemperaturesByCity.csv')\n",
    "\n",
    "# Filter by united states since others are meaningless\n",
    "temps_df = temps_df.filter(temps_df.Country == 'United States')\n",
    "\n",
    "# Drop redundant columns\n",
    "temps_df = temps_df.drop(\"dt\",\"Country\",\"Latitude\",\"Longitude\")\n",
    "\n",
    "final_temperature_df = temps_df.select(col('City'),col('AverageTemperature'),col('AverageTemperatureUncertainty')) \\\n",
    "                        .withColumnRenamed('City','city').withColumnRenamed('AverageTemperature','avg_temp') \\\n",
    "                        .withColumnRenamed('AverageTemperatureUncertainty','avg_temp_uncertainty')\n",
    "\n",
    "avg_df = final_temperature_df.select(col('city'),col('avg_temp')) \\\n",
    "            .groupBy('city').avg('avg_temp') # average_temp_df\n",
    "\n",
    "avg_uncer_df = final_temperature_df.select(col('city'),col('avg_temp_uncertainty')) \\\n",
    "            .groupBy('city').avg('avg_temp_uncertainty') # average_uncertainty_df\n",
    "\n",
    "# Now finalize...\n",
    "final_uni_df = avg_df.join(avg_uncer_df, ['city']) # join column name is same('city'), so we pass it as a list\n",
    "# Fix column names and order\n",
    "final_uni_df = final_uni_df.select(final_uni_df.city, final_uni_df['avg(avg_temp)'] \\\n",
    "                                   ,final_uni_df['avg(avg_temp_uncertainty)'])\n",
    "final_uni_df = final_uni_df.withColumnRenamed('avg(avg_temp)', 'avg_temp') \\\n",
    "                            .withColumnRenamed('avg(avg_temp_uncertainty)','avg_temp_uncertainty')\n",
    "final_uni_df.printSchema()\n",
    "final_uni_df.show(5)\n",
    "final_uni_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Step 5: Complete Project Write Up\n",
    "* Clearly state the rationale for the choice of tools and technologies for the project.\n",
    "* Propose how often the data should be updated and why."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[timestamp: double, year: int, month: int, day: int, week: int, weekday: int]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Read cities\n",
    "city_df = spark.read.options(header=\"true\",inferSchema=\"true\",nullValue = \"NULL\",delimiter=\";\").csv('us-cities-demographics.csv')\n",
    "# drop race and count columns\n",
    "city_df = city_df.drop('Race','Count')\n",
    "# Get non-null state code and city\n",
    "city_df = city_df.filter(city_df['State Code'].isNotNull() & city_df['City'].isNotNull())\n",
    "\n",
    "# Read Ports\n",
    "ports_df = spark.read.options(header=\"true\",inferSchema=\"true\",nullValue = \"NULL\").csv('airport-codes_csv.csv')\n",
    "# Lets take iso_country US, non-null iata code and non-closed records\n",
    "ports_df = ports_df.filter((ports_df.iata_code.isNotNull()) \\\n",
    "                            & (ports_df.iso_country == 'US') \\\n",
    "                            & (ports_df.type != 'closed') )\n",
    "drop_cols = ('iso_country','gps_code','elevation_ft','local_code','coordinates')\n",
    "ports_df = ports_df.drop(*drop_cols)\n",
    "\n",
    "\n",
    "# Join on 'City.city == ports_df.municipality\n",
    "combined_df = city_df.join(ports_df, city_df.City == ports_df.municipality).dropDuplicates()\n",
    "\n",
    "# Register split function\n",
    "reg_split_iso_region = F.udf(lambda iso_reg: split_iso_region(iso_reg))\n",
    "\n",
    "final_ports_df = combined_df.select(['ident','name','municipality','type','iata_code','iso_region']) \\\n",
    "                        .withColumnRenamed('ident','port_id') \\\n",
    "                        .withColumnRenamed('municipality','city') \n",
    "\n",
    "# Apply split function to iso_region\n",
    "final_ports_df = final_ports_df.withColumn('iso_region',reg_split_iso_region('iso_region').alias('region'))\n",
    "\n",
    "# Cache it for optimization\n",
    "final_ports_df = final_ports_df.cache()\n",
    "\n",
    "\n",
    "final_cities_df = combined_df.select(['City','State','Median Age','Male Population','Female Population','Total Population' \\\n",
    "                                  ,'Number of Veterans','Foreign-born','Average Household Size']) \\\n",
    "                        .withColumnRenamed('City','city') \\\n",
    "                        .withColumnRenamed('State','state') \\\n",
    "                        .withColumnRenamed('Median Age','median_age') \\\n",
    "                        .withColumnRenamed('Male Population','male_pop') \\\n",
    "                        .withColumnRenamed('Female Population','female_pop') \\\n",
    "                        .withColumnRenamed('Total Population','total_pop') \\\n",
    "                        .withColumnRenamed('Number of Veterans','veterans') \\\n",
    "                        .withColumnRenamed('Foreign-born','foreign_born') \\\n",
    "                        .withColumnRenamed('Average Household Size','avg_household_size')\n",
    "\n",
    "# Cache it for optimization\n",
    "final_cities_df = final_ports_df.cache()\n",
    "\n",
    "# Start working on immigrations data\n",
    "\n",
    "# These columns are either meaningless for the scope of the project or almost empty, so we are removing them\n",
    "drop_cols = (\"i94yr\",\"i94mon\",\"i94res\",\"count\",\"visapost\",\"occup\",\"entdepa\",\"entdepd\",\"entdepu\",\"matflag\" \\\n",
    "         ,\"biryear\",\"insnum\",\"fltno\",\"dtadfile\",\"dtaddto\",\"airline\",\"admnum\")\n",
    "\n",
    "# Month name list\n",
    "months = ['jan','feb','mar','apr','may','jun','jul','aug','sep','oct','nov','dec']\n",
    "\n",
    "#for month in months:\n",
    "immigration_data_path = f'../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat'\n",
    "\n",
    "# Read data using month in input file name\n",
    "# Jan\n",
    "jan_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_jan16_sub.sas7bdat')\n",
    "jan_df = jan_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# Feb\n",
    "feb_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_feb16_sub.sas7bdat')\n",
    "feb_df = feb_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# mar\n",
    "mar_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_mar16_sub.sas7bdat')\n",
    "mar_df = mar_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# apr\n",
    "apr_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat')\n",
    "apr_df = apr_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# may\n",
    "may_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_may16_sub.sas7bdat')\n",
    "may_df = may_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# jun\n",
    "jun_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_jun16_sub.sas7bdat')\n",
    "jun_df = jun_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# jul\n",
    "jul_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_jul16_sub.sas7bdat')\n",
    "jul_df = jul_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# aug\n",
    "aug_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_aug16_sub.sas7bdat')\n",
    "aug_df = aug_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# sep\n",
    "sep_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_sep16_sub.sas7bdat')\n",
    "sep_df = sep_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# oct\n",
    "oct_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_oct16_sub.sas7bdat')\n",
    "oct_df = oct_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# nov\n",
    "nov_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_nov16_sub.sas7bdat')\n",
    "nov_df = nov_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# dec\n",
    "dec_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_dec16_sub.sas7bdat')\n",
    "dec_df = dec_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "\n",
    "\n",
    "# Cache before using it\n",
    "df = df.cache()\n",
    "\n",
    "df = df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "\n",
    "# Fix column names and types\n",
    "df = df.withColumnRenamed('cicid','immigration_id')\n",
    "df = df.withColumn('immigration_id',df['immigration_id'].cast('int'))\n",
    "df = df.withColumnRenamed('i94cit','origin')\n",
    "df = df.withColumn('origin',df['origin'].cast('int'))\n",
    "df = df.withColumnRenamed('i94bir','age')\n",
    "df = df.withColumn('age',df['age'].cast('int'))\n",
    "df = df.withColumnRenamed('i94mode','arrival_mode')\n",
    "df = df.withColumn('arrival_mode',df['arrival_mode'].cast('int'))\n",
    "df = df.withColumnRenamed('i94visa','visa')\n",
    "df = df.withColumn('visa',df['visa'].cast('int'))\n",
    "df = df.withColumnRenamed('i94port','landing_port') \\\n",
    "    .withColumnRenamed('arrdate','arrival_date') \\\n",
    "    .withColumnRenamed('i94mode','arrival_mode') \\\n",
    "    .withColumnRenamed('i94addr','state') \\\n",
    "    .withColumnRenamed('depdate','departure_date')\n",
    "\n",
    "# Now generate time df from current immigration data using 'arrival_date' and 'departure_date' columns\n",
    "time_df = generate_time_df(df)\n",
    "time_df\n",
    "#print(f'Month: {month}, DF row count: {time_df.count()}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Let's see some data distribution"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Time data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|year|count|\n",
      "+----+-----+\n",
      "|2016|  201|\n",
      "|2015|    9|\n",
      "|2012|    2|\n",
      "|2014|    2|\n",
      "|2069|    1|\n",
      "|2020|    1|\n",
      "|2001|    1|\n",
      "|2084|    1|\n",
      "+----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "time_df.groupBy('year').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|month|count|\n",
      "+-----+-----+\n",
      "|    5|   37|\n",
      "|    4|   34|\n",
      "|    6|   32|\n",
      "|    7|   32|\n",
      "|    8|   31|\n",
      "|    9|   19|\n",
      "|    3|   19|\n",
      "|    1|    9|\n",
      "|   12|    2|\n",
      "|    2|    2|\n",
      "|   10|    1|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "time_df.groupBy('month').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n",
      "|day|count|\n",
      "+---+-----+\n",
      "|  3|   10|\n",
      "|  5|   10|\n",
      "| 15|    9|\n",
      "| 18|    9|\n",
      "| 10|    8|\n",
      "| 14|    8|\n",
      "| 13|    8|\n",
      "|  7|    8|\n",
      "|  4|    8|\n",
      "|  8|    8|\n",
      "| 12|    8|\n",
      "| 30|    7|\n",
      "| 26|    7|\n",
      "|  2|    7|\n",
      "| 17|    7|\n",
      "|  6|    7|\n",
      "| 22|    7|\n",
      "| 19|    7|\n",
      "| 16|    7|\n",
      "| 11|    7|\n",
      "+---+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "time_df.groupBy('day').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Immigrations data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+\n",
      "|state| count|\n",
      "+-----+------+\n",
      "|   FL|516283|\n",
      "|   NY|465335|\n",
      "|   CA|393687|\n",
      "|   HI|138868|\n",
      "|   TX|108269|\n",
      "|   NV|102373|\n",
      "|   GU| 86276|\n",
      "|   NJ| 64903|\n",
      "|   IL| 64794|\n",
      "|   MA| 54230|\n",
      "|   WA| 44594|\n",
      "|   GA| 30901|\n",
      "|   VA| 26037|\n",
      "|   PA| 25736|\n",
      "|   MI| 24782|\n",
      "|   DC| 23671|\n",
      "|   MD| 21258|\n",
      "|   NC| 19471|\n",
      "|   NE| 19130|\n",
      "|   LA| 17705|\n",
      "+-----+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('state').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "|visa|  count|\n",
      "+----+-------+\n",
      "|   2|2020890|\n",
      "|   1| 383181|\n",
      "|   3|  27632|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('visa').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------+\n",
      "|visatype|  count|\n",
      "+--------+-------+\n",
      "|      WT|1027119|\n",
      "|      B2| 924936|\n",
      "|      WB| 182475|\n",
      "|      B1| 179414|\n",
      "|     GMT|  67921|\n",
      "|      F1|  25285|\n",
      "|      E2|  15095|\n",
      "|      E1|   2979|\n",
      "|       I|   2887|\n",
      "|      F2|   1638|\n",
      "|      CP|    904|\n",
      "|      M1|    679|\n",
      "|      I1|    204|\n",
      "|     GMB|    127|\n",
      "|      M2|     30|\n",
      "|     CPL|      8|\n",
      "|     SBP|      2|\n",
      "+--------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('visatype').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------+\n",
      "|arrival_mode|  count|\n",
      "+------------+-------+\n",
      "|           1|2378319|\n",
      "|           3|  43401|\n",
      "|           2|   7927|\n",
      "|           9|   2056|\n",
      "+------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('arrival_mode').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n",
      "|age|count|\n",
      "+---+-----+\n",
      "| 30|57816|\n",
      "| 31|56313|\n",
      "| 33|56100|\n",
      "| 34|55999|\n",
      "| 32|55629|\n",
      "| 35|55205|\n",
      "| 29|54002|\n",
      "| 36|53915|\n",
      "| 40|52969|\n",
      "| 37|52878|\n",
      "| 28|52087|\n",
      "| 38|50706|\n",
      "| 39|49727|\n",
      "| 41|49429|\n",
      "| 42|48813|\n",
      "| 44|48488|\n",
      "| 45|48374|\n",
      "| 43|48064|\n",
      "| 27|47868|\n",
      "| 46|46334|\n",
      "+---+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('age').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Age seems distributed 'well' :)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|origin| count|\n",
      "+------+------+\n",
      "|   135|270066|\n",
      "|   209|159489|\n",
      "|   582|151434|\n",
      "|   111|149476|\n",
      "|   245|144744|\n",
      "|   148|113654|\n",
      "|   689|110979|\n",
      "|   254|110165|\n",
      "|   213| 81499|\n",
      "|   438| 76587|\n",
      "|   687| 63052|\n",
      "|   123| 59081|\n",
      "|   117| 56265|\n",
      "|   129| 47507|\n",
      "|   691| 45869|\n",
      "|   692| 39018|\n",
      "|   130| 36283|\n",
      "|   696| 35434|\n",
      "|   252| 33742|\n",
      "|   251| 31547|\n",
      "+------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('origin').count().orderBy('count', ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'df' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-8-f95505adf9e8>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;31m# Join conditions as a list\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0mconditions\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0mdf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstate\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mfinal_ports_df\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0miso_region\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlanding_port\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mfinal_ports_df\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0miata_code\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m: name 'df' is not defined"
     ]
    }
   ],
   "source": [
    "# Join conditions as a list\n",
    "conditions = [df.state == final_ports_df.iso_region, df.landing_port == final_ports_df.iata_code]\n",
    "joined_immig = df.join(final_ports_df, conditions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count jan: 1985026\n",
      "Row count feb: 1888754\n",
      "Row count mar: 2523391\n",
      "Row count apr: 2431703\n",
      "Row count may: 2634982\n",
      "Row count jun: 2614153\n",
      "Row count jul: 3368741\n",
      "Row count aug: 3081796\n",
      "Row count sep: 3303289\n",
      "Row count oct: 2755589\n",
      "Row count nov: 2358098\n",
      "Row count dec: 3004757\n"
     ]
    }
   ],
   "source": [
    "# Read data using month in input file name\n",
    "# Jan\n",
    "jan_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_jan16_sub.sas7bdat')\n",
    "jan_df = jan_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# Feb\n",
    "feb_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_feb16_sub.sas7bdat')\n",
    "feb_df = feb_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# mar\n",
    "mar_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_mar16_sub.sas7bdat')\n",
    "mar_df = mar_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# apr\n",
    "apr_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat')\n",
    "apr_df = apr_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# may\n",
    "may_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_may16_sub.sas7bdat')\n",
    "may_df = may_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# jun\n",
    "jun_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_jun16_sub.sas7bdat')\n",
    "jun_df = jun_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# jul\n",
    "jul_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_jul16_sub.sas7bdat')\n",
    "jul_df = jul_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# aug\n",
    "aug_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_aug16_sub.sas7bdat')\n",
    "aug_df = aug_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# sep\n",
    "sep_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_sep16_sub.sas7bdat')\n",
    "sep_df = sep_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# oct\n",
    "oct_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_oct16_sub.sas7bdat')\n",
    "oct_df = oct_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# nov\n",
    "nov_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_nov16_sub.sas7bdat')\n",
    "nov_df = nov_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "# dec\n",
    "dec_df = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_dec16_sub.sas7bdat')\n",
    "dec_df = dec_df.dropDuplicates().drop(*drop_cols).na.drop()\n",
    "\n",
    "print(f'Row count jan: {jan_df.count()}')\n",
    "print(f'Row count feb: {feb_df.count()}')\n",
    "print(f'Row count mar: {mar_df.count()}')\n",
    "print(f'Row count apr: {apr_df.count()}')\n",
    "print(f'Row count may: {may_df.count()}')\n",
    "print(f'Row count jun: {jun_df.count()}')\n",
    "print(f'Row count jul: {jul_df.count()}')\n",
    "print(f'Row count aug: {aug_df.count()}')\n",
    "print(f'Row count sep: {sep_df.count()}')\n",
    "print(f'Row count oct: {oct_df.count()}')\n",
    "print(f'Row count nov: {nov_df.count()}')\n",
    "print(f'Row count dec: {dec_df.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count jan: 534975\n"
     ]
    }
   ],
   "source": [
    "conditions = [jan_df.i94addr == final_ports_df.iso_region, jan_df.i94port == final_ports_df.iata_code]\n",
    "join_jan = jan_df.join(final_ports_df, conditions)\n",
    "print(f'Row count jan: {join_jan.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count feb: 466855\n"
     ]
    }
   ],
   "source": [
    "conditions = [feb_df.i94addr == final_ports_df.iso_region, feb_df.i94port == final_ports_df.iata_code]\n",
    "join_feb = feb_df.join(final_ports_df, conditions)\n",
    "print(f'Row count feb: {join_feb.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count mar: 623297\n"
     ]
    }
   ],
   "source": [
    "conditions = [mar_df.i94addr == final_ports_df.iso_region, mar_df.i94port == final_ports_df.iata_code]\n",
    "join_mar = mar_df.join(final_ports_df, conditions)\n",
    "print(f'Row count mar: {join_mar.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count apr: 481368\n"
     ]
    }
   ],
   "source": [
    "conditions = [apr_df.i94addr == final_ports_df.iso_region, apr_df.i94port == final_ports_df.iata_code]\n",
    "join_apr = apr_df.join(final_ports_df, conditions)\n",
    "print(f'Row count apr: {join_apr.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count may: 614628\n"
     ]
    }
   ],
   "source": [
    "conditions = [may_df.i94addr == final_ports_df.iso_region, may_df.i94port == final_ports_df.iata_code]\n",
    "join_may = may_df.join(final_ports_df, conditions)\n",
    "print(f'Row count may: {join_may.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count jun: 600369\n"
     ]
    }
   ],
   "source": [
    "conditions = [jun_df.i94addr == final_ports_df.iso_region, jun_df.i94port == final_ports_df.iata_code]\n",
    "join_jun = jun_df.join(final_ports_df, conditions)\n",
    "print(f'Row count jun: {join_jun.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count jul: 816280\n"
     ]
    }
   ],
   "source": [
    "conditions = [jul_df.i94addr == final_ports_df.iso_region, jul_df.i94port == final_ports_df.iata_code]\n",
    "join_jul = jul_df.join(final_ports_df, conditions)\n",
    "print(f'Row count jul: {join_jul.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count aug: 724842\n"
     ]
    }
   ],
   "source": [
    "conditions = [aug_df.i94addr == final_ports_df.iso_region, aug_df.i94port == final_ports_df.iata_code]\n",
    "join_aug = aug_df.join(final_ports_df, conditions)\n",
    "print(f'Row count aug: {join_aug.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count sep: 723015\n"
     ]
    }
   ],
   "source": [
    "conditions = [sep_df.i94addr == final_ports_df.iso_region, sep_df.i94port == final_ports_df.iata_code]\n",
    "join_sep = sep_df.join(final_ports_df, conditions)\n",
    "print(f'Row count sep: {join_sep.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count oct: 645835\n"
     ]
    }
   ],
   "source": [
    "conditions = [oct_df.i94addr == final_ports_df.iso_region, oct_df.i94port == final_ports_df.iata_code]\n",
    "join_oct = oct_df.join(final_ports_df, conditions)\n",
    "print(f'Row count oct: {join_oct.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count nov: 609114\n"
     ]
    }
   ],
   "source": [
    "conditions = [nov_df.i94addr == final_ports_df.iso_region, nov_df.i94port == final_ports_df.iata_code]\n",
    "join_nov = nov_df.join(final_ports_df, conditions)\n",
    "print(f'Row count nov: {join_nov.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row count dec: 824109\n"
     ]
    }
   ],
   "source": [
    "conditions = [dec_df.i94addr == final_ports_df.iso_region, dec_df.i94port == final_ports_df.iata_code]\n",
    "join_dec = dec_df.join(final_ports_df, conditions)\n",
    "print(f'Row count dec: {join_dec.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cicid: double (nullable = true)\n",
      " |-- i94cit: double (nullable = true)\n",
      " |-- i94port: string (nullable = true)\n",
      " |-- arrdate: double (nullable = true)\n",
      " |-- i94mode: double (nullable = true)\n",
      " |-- i94addr: string (nullable = true)\n",
      " |-- depdate: double (nullable = true)\n",
      " |-- i94bir: double (nullable = true)\n",
      " |-- i94visa: double (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- visatype: string (nullable = true)\n",
      " |-- port_id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- iata_code: string (nullable = true)\n",
      " |-- iso_region: string (nullable = true)\n",
      "\n",
      "Big df count: 1001830\n"
     ]
    }
   ],
   "source": [
    "big_df = join_jan.union(join_feb)\n",
    "big_df.printSchema()\n",
    "print(f'Big df count: {big_df.count()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1001718"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "big_df.dropDuplicates().na.drop().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": 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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}