Capstone Project Template.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Study of Immigration Data in the United States\n",
    "### Data Engineering Capstone Project\n",
    "\n",
    "#### Project Summary\n",
    "\n",
    "In this project, we will be looking at the immigration data for the united states. More specifically, we're interested in looking at the following phenomena:\n",
    "* the effects of temperature on the volume of travellers, \n",
    "* the seasonality of travel \n",
    "* the connection between the volume of travel and the number of entry ports (ie airports) \n",
    "* the connection between the volume of travel and the demographics of various cities\n",
    "\n",
    "To accomplish this study, we will be using the following datasets:\n",
    "\n",
    "* **I94 Immigration Data**: This data comes from the US National Tourism and Trade Office and includes the contents of the i94 form on entry to the united states. A data dictionary is included in the workspace.\n",
    "    * _countries.csv_ : table containing country codes used in the dataset, extracted from the data dictionary\n",
    "    * _i94portCodes.csv_: table containing city codes used in the dataset, extracted from the data dictionary\n",
    "\n",
    "* **World Temperature Data**: This dataset comes from Kaggle and includes the temperatures of various cities in the world fomr 1743 to 2013.\n",
    "* **U.S. City Demographic Data**: This data comes from OpenSoft. It contains information about the demographics of all US cities and census-designated places with a population greater or equal to 65,000 and comes from the US Census Bureau's 2015 American Community Survey.\n",
    "* **Airport Code Table**: This is a simple table of airport codes and corresponding cities.\n",
    "\n",
    "\n",
    "In order to accomplish this, we will aggregate our data as follows:\n",
    "* aggregate based on time (year, month, day, etc...) \n",
    "* aggregate data by cities and airports\n",
    "* look at the impact of temperatures on the in and ouflux of travelers\n",
    "* the impact on regional demographics\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",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import udf, date_add\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql.functions import desc\n",
    "from pyspark.sql.functions import asc\n",
    "from pyspark.sql.functions import sum as Fsum\n",
    "\n",
    "import datetime\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd"
   ]
  },
  {
   "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",
    "#### 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": [
    "#### Immigration Dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The immigration dataset is rather large, containing approximately 3m lines. We have will use a subset of approx 1000 rows in a csv to explore it"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Read in the data here\n",
    "df_immig_sample = pd.read_csv('immigration_data_sample.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "This data set contains details about traveler entering the united states."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Index(['Unnamed: 0', 'cicid', 'i94yr', 'i94mon', 'i94cit', 'i94res', 'i94port',\n",
       "       'arrdate', 'i94mode', 'i94addr', 'depdate', 'i94bir', 'i94visa',\n",
       "       'count', 'dtadfile', 'visapost', 'occup', 'entdepa', 'entdepd',\n",
       "       'entdepu', 'matflag', 'biryear', 'dtaddto', 'gender', 'insnum',\n",
       "       'airline', 'admnum', 'fltno', 'visatype'],\n",
       "      dtype='object')"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_immig_sample.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The definition of the fiels is included in the file **I94_SAS_Labels_Descriptions.SAS**\n",
    "We'll primarily be interested in the following fields:\n",
    "* i94cit : country of citizenship\n",
    "* i94res : country of residence\n",
    "* i94port: arrival airport\n",
    "* arrdate: arrival date. \n",
    "* i94mode\n",
    "* i94addr\n",
    "* depdate\n",
    "* i94bir\n",
    "* i94visa\n",
    "* occup\n",
    "* biryear\n",
    "* dtaddto\n",
    "* gender\n",
    "* insnum\n",
    "* airline\n",
    "* admnum\n",
    "* fltno\n",
    "* visatype\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's increase the number of columns that can be displayed at once to have a better look at the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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>Unnamed: 0</th>\n",
       "      <th>cicid</th>\n",
       "      <th>i94yr</th>\n",
       "      <th>i94mon</th>\n",
       "      <th>i94cit</th>\n",
       "      <th>i94res</th>\n",
       "      <th>i94port</th>\n",
       "      <th>arrdate</th>\n",
       "      <th>i94mode</th>\n",
       "      <th>i94addr</th>\n",
       "      <th>depdate</th>\n",
       "      <th>i94bir</th>\n",
       "      <th>i94visa</th>\n",
       "      <th>count</th>\n",
       "      <th>dtadfile</th>\n",
       "      <th>visapost</th>\n",
       "      <th>occup</th>\n",
       "      <th>entdepa</th>\n",
       "      <th>entdepd</th>\n",
       "      <th>entdepu</th>\n",
       "      <th>matflag</th>\n",
       "      <th>biryear</th>\n",
       "      <th>dtaddto</th>\n",
       "      <th>gender</th>\n",
       "      <th>insnum</th>\n",
       "      <th>airline</th>\n",
       "      <th>admnum</th>\n",
       "      <th>fltno</th>\n",
       "      <th>visatype</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2027561</td>\n",
       "      <td>4084316.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>209.0</td>\n",
       "      <td>209.0</td>\n",
       "      <td>HHW</td>\n",
       "      <td>20566.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>HI</td>\n",
       "      <td>20573.0</td>\n",
       "      <td>61.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160422</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>G</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1955.0</td>\n",
       "      <td>07202016</td>\n",
       "      <td>F</td>\n",
       "      <td>NaN</td>\n",
       "      <td>JL</td>\n",
       "      <td>5.658267e+10</td>\n",
       "      <td>00782</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2171295</td>\n",
       "      <td>4422636.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>582.0</td>\n",
       "      <td>582.0</td>\n",
       "      <td>MCA</td>\n",
       "      <td>20567.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>TX</td>\n",
       "      <td>20568.0</td>\n",
       "      <td>26.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160423</td>\n",
       "      <td>MTR</td>\n",
       "      <td>NaN</td>\n",
       "      <td>G</td>\n",
       "      <td>R</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1990.0</td>\n",
       "      <td>10222016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>*GA</td>\n",
       "      <td>9.436200e+10</td>\n",
       "      <td>XBLNG</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>589494</td>\n",
       "      <td>1195600.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>148.0</td>\n",
       "      <td>112.0</td>\n",
       "      <td>OGG</td>\n",
       "      <td>20551.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>FL</td>\n",
       "      <td>20571.0</td>\n",
       "      <td>76.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160407</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>G</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1940.0</td>\n",
       "      <td>07052016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>LH</td>\n",
       "      <td>5.578047e+10</td>\n",
       "      <td>00464</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2631158</td>\n",
       "      <td>5291768.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>297.0</td>\n",
       "      <td>297.0</td>\n",
       "      <td>LOS</td>\n",
       "      <td>20572.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>CA</td>\n",
       "      <td>20581.0</td>\n",
       "      <td>25.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160428</td>\n",
       "      <td>DOH</td>\n",
       "      <td>NaN</td>\n",
       "      <td>G</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1991.0</td>\n",
       "      <td>10272016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>QR</td>\n",
       "      <td>9.478970e+10</td>\n",
       "      <td>00739</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>3032257</td>\n",
       "      <td>985523.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>111.0</td>\n",
       "      <td>111.0</td>\n",
       "      <td>CHM</td>\n",
       "      <td>20550.0</td>\n",
       "      <td>3.0</td>\n",
       "      <td>NY</td>\n",
       "      <td>20553.0</td>\n",
       "      <td>19.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160406</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Z</td>\n",
       "      <td>K</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1997.0</td>\n",
       "      <td>07042016</td>\n",
       "      <td>F</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4.232257e+10</td>\n",
       "      <td>LAND</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>721257</td>\n",
       "      <td>1481650.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>577.0</td>\n",
       "      <td>577.0</td>\n",
       "      <td>ATL</td>\n",
       "      <td>20552.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>GA</td>\n",
       "      <td>20606.0</td>\n",
       "      <td>51.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160408</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>T</td>\n",
       "      <td>N</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1965.0</td>\n",
       "      <td>10072016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>DL</td>\n",
       "      <td>7.368526e+08</td>\n",
       "      <td>910</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>1072780</td>\n",
       "      <td>2197173.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>245.0</td>\n",
       "      <td>245.0</td>\n",
       "      <td>SFR</td>\n",
       "      <td>20556.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>CA</td>\n",
       "      <td>20635.0</td>\n",
       "      <td>48.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160412</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>T</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1968.0</td>\n",
       "      <td>10112016</td>\n",
       "      <td>F</td>\n",
       "      <td>NaN</td>\n",
       "      <td>CX</td>\n",
       "      <td>7.863122e+08</td>\n",
       "      <td>870</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>112205</td>\n",
       "      <td>232708.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>113.0</td>\n",
       "      <td>135.0</td>\n",
       "      <td>NYC</td>\n",
       "      <td>20546.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NY</td>\n",
       "      <td>20554.0</td>\n",
       "      <td>33.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160402</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>G</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1983.0</td>\n",
       "      <td>06302016</td>\n",
       "      <td>F</td>\n",
       "      <td>NaN</td>\n",
       "      <td>BA</td>\n",
       "      <td>5.547449e+10</td>\n",
       "      <td>00117</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>2577162</td>\n",
       "      <td>5227851.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>131.0</td>\n",
       "      <td>131.0</td>\n",
       "      <td>CHI</td>\n",
       "      <td>20572.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>IL</td>\n",
       "      <td>20575.0</td>\n",
       "      <td>39.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160428</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>O</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1977.0</td>\n",
       "      <td>07262016</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>LX</td>\n",
       "      <td>5.941342e+10</td>\n",
       "      <td>00008</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>10930</td>\n",
       "      <td>13213.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>116.0</td>\n",
       "      <td>116.0</td>\n",
       "      <td>LOS</td>\n",
       "      <td>20545.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>CA</td>\n",
       "      <td>20553.0</td>\n",
       "      <td>35.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20160401</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>O</td>\n",
       "      <td>O</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1981.0</td>\n",
       "      <td>06292016</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>AA</td>\n",
       "      <td>5.544979e+10</td>\n",
       "      <td>00109</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   Unnamed: 0      cicid   i94yr  i94mon  i94cit  i94res i94port  arrdate  \\\n",
       "0     2027561  4084316.0  2016.0     4.0   209.0   209.0     HHW  20566.0   \n",
       "1     2171295  4422636.0  2016.0     4.0   582.0   582.0     MCA  20567.0   \n",
       "2      589494  1195600.0  2016.0     4.0   148.0   112.0     OGG  20551.0   \n",
       "3     2631158  5291768.0  2016.0     4.0   297.0   297.0     LOS  20572.0   \n",
       "4     3032257   985523.0  2016.0     4.0   111.0   111.0     CHM  20550.0   \n",
       "5      721257  1481650.0  2016.0     4.0   577.0   577.0     ATL  20552.0   \n",
       "6     1072780  2197173.0  2016.0     4.0   245.0   245.0     SFR  20556.0   \n",
       "7      112205   232708.0  2016.0     4.0   113.0   135.0     NYC  20546.0   \n",
       "8     2577162  5227851.0  2016.0     4.0   131.0   131.0     CHI  20572.0   \n",
       "9       10930    13213.0  2016.0     4.0   116.0   116.0     LOS  20545.0   \n",
       "\n",
       "   i94mode i94addr  depdate  i94bir  i94visa  count  dtadfile visapost occup  \\\n",
       "0      1.0      HI  20573.0    61.0      2.0    1.0  20160422      NaN   NaN   \n",
       "1      1.0      TX  20568.0    26.0      2.0    1.0  20160423      MTR   NaN   \n",
       "2      1.0      FL  20571.0    76.0      2.0    1.0  20160407      NaN   NaN   \n",
       "3      1.0      CA  20581.0    25.0      2.0    1.0  20160428      DOH   NaN   \n",
       "4      3.0      NY  20553.0    19.0      2.0    1.0  20160406      NaN   NaN   \n",
       "5      1.0      GA  20606.0    51.0      2.0    1.0  20160408      NaN   NaN   \n",
       "6      1.0      CA  20635.0    48.0      2.0    1.0  20160412      NaN   NaN   \n",
       "7      1.0      NY  20554.0    33.0      2.0    1.0  20160402      NaN   NaN   \n",
       "8      1.0      IL  20575.0    39.0      2.0    1.0  20160428      NaN   NaN   \n",
       "9      1.0      CA  20553.0    35.0      2.0    1.0  20160401      NaN   NaN   \n",
       "\n",
       "  entdepa entdepd  entdepu matflag  biryear   dtaddto gender  insnum airline  \\\n",
       "0       G       O      NaN       M   1955.0  07202016      F     NaN      JL   \n",
       "1       G       R      NaN       M   1990.0  10222016      M     NaN     *GA   \n",
       "2       G       O      NaN       M   1940.0  07052016      M     NaN      LH   \n",
       "3       G       O      NaN       M   1991.0  10272016      M     NaN      QR   \n",
       "4       Z       K      NaN       M   1997.0  07042016      F     NaN     NaN   \n",
       "5       T       N      NaN       M   1965.0  10072016      M     NaN      DL   \n",
       "6       T       O      NaN       M   1968.0  10112016      F     NaN      CX   \n",
       "7       G       O      NaN       M   1983.0  06302016      F     NaN      BA   \n",
       "8       O       O      NaN       M   1977.0  07262016    NaN     NaN      LX   \n",
       "9       O       O      NaN       M   1981.0  06292016    NaN     NaN      AA   \n",
       "\n",
       "         admnum  fltno visatype  \n",
       "0  5.658267e+10  00782       WT  \n",
       "1  9.436200e+10  XBLNG       B2  \n",
       "2  5.578047e+10  00464       WT  \n",
       "3  9.478970e+10  00739       B2  \n",
       "4  4.232257e+10   LAND       WT  \n",
       "5  7.368526e+08    910       B2  \n",
       "6  7.863122e+08    870       B2  \n",
       "7  5.547449e+10  00117       WT  \n",
       "8  5.941342e+10  00008       WT  \n",
       "9  5.544979e+10  00109       WT  "
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.set_option('display.max_columns', 50)\n",
    "df_immig_sample.head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Adding data dictionaries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "In the next step, we add details from the data dictionnary. These will be replace the codes in our data model since we work with denormalized data models."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "First we add the dictionary for columns I94CIT & I94RES (countries.csv) which we assume corresponds to country of citizenship and country of residence of the travelers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_countryCodes = pd.read_csv('countries.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(289, 2)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_countryCodes.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "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>code</th>\n",
       "      <th>country</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>582</td>\n",
       "      <td>MEXICO</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>236</td>\n",
       "      <td>AFGHANISTAN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>101</td>\n",
       "      <td>ALBANIA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>316</td>\n",
       "      <td>ALGERIA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>102</td>\n",
       "      <td>ANDORRA</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   code      country\n",
       "0   582      MEXICO \n",
       "1   236  AFGHANISTAN\n",
       "2   101      ALBANIA\n",
       "3   316      ALGERIA\n",
       "4   102      ANDORRA"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_countryCodes.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Next, we add a correspondance table between i94port codes and city that was the port of entry (i94portCodes.csv)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "i94portCodes = pd.read_csv('i94portCodes.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(660, 3)"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "i94portCodes.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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>code</th>\n",
       "      <th>location</th>\n",
       "      <th>state</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ALC</td>\n",
       "      <td>ALCAN</td>\n",
       "      <td>AK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>ANC</td>\n",
       "      <td>ANCHORAGE</td>\n",
       "      <td>AK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>BAR</td>\n",
       "      <td>BAKER AAF - BAKER ISLAND</td>\n",
       "      <td>AK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>DAC</td>\n",
       "      <td>DALTONS CACHE</td>\n",
       "      <td>AK</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>PIZ</td>\n",
       "      <td>DEW STATION PT LAY DEW</td>\n",
       "      <td>AK</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  code                  location state\n",
       "0  ALC                     ALCAN    AK\n",
       "1  ANC                 ANCHORAGE    AK\n",
       "2  BAR  BAKER AAF - BAKER ISLAND    AK\n",
       "3  DAC             DALTONS CACHE    AK\n",
       "4  PIZ    DEW STATION PT LAY DEW    AK"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "i94portCodes.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Demographic data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Our dataset contains details on the reason for traveling. It might be interesting to see if there is a connection between the flow of immigration and the demographic data of various US cities.\n",
    "\n",
    "Let's load the cities demographics\n",
    "The separator used in the csv is a ';' rather than a ','"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Read in the data here\n",
    "df_demographics = pd.read_csv('us-cities-demographics.csv', sep=';')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(2891, 12)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "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>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",
       "      <th>Race</th>\n",
       "      <th>Count</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.0</td>\n",
       "      <td>41862.0</td>\n",
       "      <td>82463</td>\n",
       "      <td>1562.0</td>\n",
       "      <td>30908.0</td>\n",
       "      <td>2.60</td>\n",
       "      <td>MD</td>\n",
       "      <td>Hispanic or Latino</td>\n",
       "      <td>25924</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.0</td>\n",
       "      <td>49500.0</td>\n",
       "      <td>93629</td>\n",
       "      <td>4147.0</td>\n",
       "      <td>32935.0</td>\n",
       "      <td>2.39</td>\n",
       "      <td>MA</td>\n",
       "      <td>White</td>\n",
       "      <td>58723</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.0</td>\n",
       "      <td>46799.0</td>\n",
       "      <td>84839</td>\n",
       "      <td>4819.0</td>\n",
       "      <td>8229.0</td>\n",
       "      <td>2.58</td>\n",
       "      <td>AL</td>\n",
       "      <td>Asian</td>\n",
       "      <td>4759</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.0</td>\n",
       "      <td>87105.0</td>\n",
       "      <td>175232</td>\n",
       "      <td>5821.0</td>\n",
       "      <td>33878.0</td>\n",
       "      <td>3.18</td>\n",
       "      <td>CA</td>\n",
       "      <td>Black or African-American</td>\n",
       "      <td>24437</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.0</td>\n",
       "      <td>143873.0</td>\n",
       "      <td>281913</td>\n",
       "      <td>5829.0</td>\n",
       "      <td>86253.0</td>\n",
       "      <td>2.73</td>\n",
       "      <td>NJ</td>\n",
       "      <td>White</td>\n",
       "      <td>76402</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.0   \n",
       "1            Quincy  Massachusetts        41.0          44129.0   \n",
       "2            Hoover        Alabama        38.5          38040.0   \n",
       "3  Rancho Cucamonga     California        34.5          88127.0   \n",
       "4            Newark     New Jersey        34.6         138040.0   \n",
       "\n",
       "   Female Population  Total Population  Number of Veterans  Foreign-born  \\\n",
       "0            41862.0             82463              1562.0       30908.0   \n",
       "1            49500.0             93629              4147.0       32935.0   \n",
       "2            46799.0             84839              4819.0        8229.0   \n",
       "3            87105.0            175232              5821.0       33878.0   \n",
       "4           143873.0            281913              5829.0       86253.0   \n",
       "\n",
       "   Average Household Size State Code                       Race  Count  \n",
       "0                    2.60         MD         Hispanic or Latino  25924  \n",
       "1                    2.39         MA                      White  58723  \n",
       "2                    2.58         AL                      Asian   4759  \n",
       "3                    3.18         CA  Black or African-American  24437  \n",
       "4                    2.73         NJ                      White  76402  "
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Index(['City', 'State', 'Median Age', 'Male Population', 'Female Population',\n",
       "       'Total Population', 'Number of Veterans', 'Foreign-born',\n",
       "       'Average Household Size', 'State Code', 'Race', 'Count'],\n",
       "      dtype='object')"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# we look at the list of available columns in the dataset\n",
    "df_demographics.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Next, we load the airport codes. This data will allow us to connect airport data to the airport codes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Airport data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since airports are the point of entry for these immigrants, we will include airport information in our data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Read in the data here\n",
    "df_airports = pd.read_csv('airport-codes_csv.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Index(['ident', 'type', 'name', 'elevation_ft', 'continent', 'iso_country',\n",
       "       'iso_region', 'municipality', 'gps_code', 'iata_code', 'local_code',\n",
       "       'coordinates'],\n",
       "      dtype='object')"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "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>ident</th>\n",
       "      <th>type</th>\n",
       "      <th>name</th>\n",
       "      <th>elevation_ft</th>\n",
       "      <th>continent</th>\n",
       "      <th>iso_country</th>\n",
       "      <th>iso_region</th>\n",
       "      <th>municipality</th>\n",
       "      <th>gps_code</th>\n",
       "      <th>iata_code</th>\n",
       "      <th>local_code</th>\n",
       "      <th>coordinates</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>00A</td>\n",
       "      <td>heliport</td>\n",
       "      <td>Total Rf Heliport</td>\n",
       "      <td>11.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-PA</td>\n",
       "      <td>Bensalem</td>\n",
       "      <td>00A</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00A</td>\n",
       "      <td>-74.93360137939453, 40.07080078125</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>00AA</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Aero B Ranch Airport</td>\n",
       "      <td>3435.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-KS</td>\n",
       "      <td>Leoti</td>\n",
       "      <td>00AA</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00AA</td>\n",
       "      <td>-101.473911, 38.704022</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>00AK</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Lowell Field</td>\n",
       "      <td>450.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Anchor Point</td>\n",
       "      <td>00AK</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00AK</td>\n",
       "      <td>-151.695999146, 59.94919968</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>00AL</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Epps Airpark</td>\n",
       "      <td>820.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-AL</td>\n",
       "      <td>Harvest</td>\n",
       "      <td>00AL</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00AL</td>\n",
       "      <td>-86.77030181884766, 34.86479949951172</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>00AR</td>\n",
       "      <td>closed</td>\n",
       "      <td>Newport Hospital &amp; Clinic Heliport</td>\n",
       "      <td>237.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-AR</td>\n",
       "      <td>Newport</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-91.254898, 35.6087</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  ident           type                                name  elevation_ft  \\\n",
       "0   00A       heliport                   Total Rf Heliport          11.0   \n",
       "1  00AA  small_airport                Aero B Ranch Airport        3435.0   \n",
       "2  00AK  small_airport                        Lowell Field         450.0   \n",
       "3  00AL  small_airport                        Epps Airpark         820.0   \n",
       "4  00AR         closed  Newport Hospital & Clinic Heliport         237.0   \n",
       "\n",
       "  continent iso_country iso_region  municipality gps_code iata_code  \\\n",
       "0       NaN          US      US-PA      Bensalem      00A       NaN   \n",
       "1       NaN          US      US-KS         Leoti     00AA       NaN   \n",
       "2       NaN          US      US-AK  Anchor Point     00AK       NaN   \n",
       "3       NaN          US      US-AL       Harvest     00AL       NaN   \n",
       "4       NaN          US      US-AR       Newport      NaN       NaN   \n",
       "\n",
       "  local_code                            coordinates  \n",
       "0        00A     -74.93360137939453, 40.07080078125  \n",
       "1       00AA                 -101.473911, 38.704022  \n",
       "2       00AK            -151.695999146, 59.94919968  \n",
       "3       00AL  -86.77030181884766, 34.86479949951172  \n",
       "4        NaN                    -91.254898, 35.6087  "
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### World Temperature data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We expect there to be a connection between climate and tourism data. Therefore, we will add the world temperature data to see if climate has any impact on the volume of tourists."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "fname = '../../data2/GlobalLandTemperaturesByCity.csv'\n",
    "df_temperature = pd.read_csv(fname)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(8599212, 7)"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "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>dt</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>AverageTemperatureUncertainty</th>\n",
       "      <th>City</th>\n",
       "      <th>Country</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1743-11-01</td>\n",
       "      <td>6.068</td>\n",
       "      <td>1.737</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1743-12-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1744-01-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1744-02-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1744-03-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           dt  AverageTemperature  AverageTemperatureUncertainty   City  \\\n",
       "0  1743-11-01               6.068                          1.737  Århus   \n",
       "1  1743-12-01                 NaN                            NaN  Århus   \n",
       "2  1744-01-01                 NaN                            NaN  Århus   \n",
       "3  1744-02-01                 NaN                            NaN  Århus   \n",
       "4  1744-03-01                 NaN                            NaN  Århus   \n",
       "\n",
       "   Country Latitude Longitude  \n",
       "0  Denmark   57.05N    10.33E  \n",
       "1  Denmark   57.05N    10.33E  \n",
       "2  Denmark   57.05N    10.33E  \n",
       "3  Denmark   57.05N    10.33E  \n",
       "4  Denmark   57.05N    10.33E  "
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Full immigration dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Finally, we load the full immigration dataset into a spark dataframe since it is quite large"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.\\\n",
    "config(\"spark.jars.packages\",\"saurfang:spark-sas7bdat:2.0.0-s_2.11\")\\\n",
    ".enableHiveSupport().getOrCreate()\n",
    "df_immigration = spark.read.format('com.github.saurfang.sas.spark').load('../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3096313"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_immigration.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cicid: double (nullable = true)\n",
      " |-- i94yr: double (nullable = true)\n",
      " |-- i94mon: double (nullable = true)\n",
      " |-- i94cit: double (nullable = true)\n",
      " |-- i94res: 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",
      " |-- count: double (nullable = true)\n",
      " |-- dtadfile: string (nullable = true)\n",
      " |-- visapost: string (nullable = true)\n",
      " |-- occup: string (nullable = true)\n",
      " |-- entdepa: string (nullable = true)\n",
      " |-- entdepd: string (nullable = true)\n",
      " |-- entdepu: string (nullable = true)\n",
      " |-- matflag: string (nullable = true)\n",
      " |-- biryear: double (nullable = true)\n",
      " |-- dtaddto: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- insnum: string (nullable = true)\n",
      " |-- airline: string (nullable = true)\n",
      " |-- admnum: double (nullable = true)\n",
      " |-- fltno: string (nullable = true)\n",
      " |-- visatype: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_immigration.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We see here that the schema is identical to the sample dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#write to parquet\n",
    "df_immigration.write.parquet(\"sas_data\")\n",
    "df_immigration=spark.read.parquet(\"sas_data\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 2: Explore and Assess the Data\n",
    "#### Explore the Data \n",
    "Identify data quality issues, like missing values, duplicate data, etc.\n",
    "\n",
    "#### Cleaning Steps\n",
    "Document steps necessary to clean the data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "##### Temperature data\n",
    "First, we start looking at the temperature dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(8599212, 7)"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's see if we need to keep all this data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "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>dt</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>AverageTemperatureUncertainty</th>\n",
       "      <th>City</th>\n",
       "      <th>Country</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1743-11-01</td>\n",
       "      <td>6.068</td>\n",
       "      <td>1.737</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1743-12-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1744-01-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1744-02-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1744-03-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           dt  AverageTemperature  AverageTemperatureUncertainty   City  \\\n",
       "0  1743-11-01               6.068                          1.737  Århus   \n",
       "1  1743-12-01                 NaN                            NaN  Århus   \n",
       "2  1744-01-01                 NaN                            NaN  Århus   \n",
       "3  1744-02-01                 NaN                            NaN  Århus   \n",
       "4  1744-03-01                 NaN                            NaN  Århus   \n",
       "\n",
       "   Country Latitude Longitude  \n",
       "0  Denmark   57.05N    10.33E  \n",
       "1  Denmark   57.05N    10.33E  \n",
       "2  Denmark   57.05N    10.33E  \n",
       "3  Denmark   57.05N    10.33E  \n",
       "4  Denmark   57.05N    10.33E  "
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "159"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature['Country'].nunique()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The dataframe contains temperature data for 159 countries since the year 1743.\n",
    "We will reduce the size of the dataset to make it more manageable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Keep only data for the United States\n",
    "df_temperature = df_temperature[df_temperature['Country']=='United States']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Convert the date to datetime objects\n",
    "df_temperature['convertedDate'] = pd.to_datetime(df_temperature.dt)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since we are focusing on air travel, we'll exclude any data prior to 1950 since commercial air travel did't develop until after the second world war in the 1950s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Remove all dates prior to 1950\n",
    "df_temperature=df_temperature[df_temperature['convertedDate']>\"1950-01-01\"].copy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(196348, 8)"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Timestamp('2013-09-01 00:00:00')"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Let's check the most recent date in the dataset\n",
    "df_temperature['convertedDate'].max()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "No temperature that's available here can be joined with our immigration dataset. We will assume for the purposes of this project that we'd have data that can be joined with our immigration dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Now, let's check for missing data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dt                               0\n",
       "AverageTemperature               1\n",
       "AverageTemperatureUncertainty    1\n",
       "City                             0\n",
       "Country                          0\n",
       "Latitude                         0\n",
       "Longitude                        0\n",
       "convertedDate                    0\n",
       "dtype: int64"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Let's check for null values.\n",
    "df_temperature.isnull().sum()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "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>dt</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>AverageTemperatureUncertainty</th>\n",
       "      <th>City</th>\n",
       "      <th>Country</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "      <th>convertedDate</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>287781</th>\n",
       "      <td>2013-09-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Anchorage</td>\n",
       "      <td>United States</td>\n",
       "      <td>61.88N</td>\n",
       "      <td>151.13W</td>\n",
       "      <td>2013-09-01</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                dt  AverageTemperature  AverageTemperatureUncertainty  \\\n",
       "287781  2013-09-01                 NaN                            NaN   \n",
       "\n",
       "             City        Country Latitude Longitude convertedDate  \n",
       "287781  Anchorage  United States   61.88N   151.13W    2013-09-01  "
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature[df_temperature.AverageTemperature.isnull()]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Normally, we would have to fix the null values present in the temperature dataset for Anchorage in September 2019. Two possible straightforward strategies for fixing this issue would be:\n",
    "\n",
    "* Average out the values of August and October 2019 for Anchorage;\n",
    "* Using an average of the historical data for the month of September for Anchorage;\n",
    "\n",
    "Option 2 while feasiable is less desirable since temperatures appear to be higher in 2013 compared to previous years and might create an outlier in the dataset.\n",
    "\n",
    "Normally our dataset would include data all the way to aprl 2016 to allow us to join the data with our immigration dataset, making it possible to use option 1.\n",
    "However, since no data is available in this set beyond 2013-09-01 and because our immigration set only covers the month of april 2016, we'll leave this missing data problem as is."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Finally, let's make sure the combination of city and date can be used as a primary key.\n",
    "We've assumed that each row represents a combination of city and date. Let's check that"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(196348, 8)"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(189472, 2)"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature[['City','convertedDate']].drop_duplicates().shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "It looks like there can be multiple entries for a give city. Let's try to look at an example of multiple entries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "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>dt</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>AverageTemperatureUncertainty</th>\n",
       "      <th>City</th>\n",
       "      <th>Country</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "      <th>convertedDate</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>405836</th>\n",
       "      <td>1950-02-01</td>\n",
       "      <td>1.655</td>\n",
       "      <td>0.057</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>39.38N</td>\n",
       "      <td>76.99W</td>\n",
       "      <td>1950-02-01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>405837</th>\n",
       "      <td>1950-03-01</td>\n",
       "      <td>3.871</td>\n",
       "      <td>0.232</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>39.38N</td>\n",
       "      <td>76.99W</td>\n",
       "      <td>1950-03-01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>405838</th>\n",
       "      <td>1950-04-01</td>\n",
       "      <td>9.678</td>\n",
       "      <td>0.191</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>39.38N</td>\n",
       "      <td>76.99W</td>\n",
       "      <td>1950-04-01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>405839</th>\n",
       "      <td>1950-05-01</td>\n",
       "      <td>16.786</td>\n",
       "      <td>0.234</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>39.38N</td>\n",
       "      <td>76.99W</td>\n",
       "      <td>1950-05-01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>405840</th>\n",
       "      <td>1950-06-01</td>\n",
       "      <td>21.548</td>\n",
       "      <td>0.222</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>39.38N</td>\n",
       "      <td>76.99W</td>\n",
       "      <td>1950-06-01</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                dt  AverageTemperature  AverageTemperatureUncertainty  \\\n",
       "405836  1950-02-01               1.655                          0.057   \n",
       "405837  1950-03-01               3.871                          0.232   \n",
       "405838  1950-04-01               9.678                          0.191   \n",
       "405839  1950-05-01              16.786                          0.234   \n",
       "405840  1950-06-01              21.548                          0.222   \n",
       "\n",
       "             City        Country Latitude Longitude convertedDate  \n",
       "405836  Arlington  United States   39.38N    76.99W    1950-02-01  \n",
       "405837  Arlington  United States   39.38N    76.99W    1950-03-01  \n",
       "405838  Arlington  United States   39.38N    76.99W    1950-04-01  \n",
       "405839  Arlington  United States   39.38N    76.99W    1950-05-01  \n",
       "405840  Arlington  United States   39.38N    76.99W    1950-06-01  "
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature[df_temperature[['City','convertedDate']].duplicated()].head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "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>dt</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>AverageTemperatureUncertainty</th>\n",
       "      <th>City</th>\n",
       "      <th>Country</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "      <th>convertedDate</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>402597</th>\n",
       "      <td>1950-02-01</td>\n",
       "      <td>11.144</td>\n",
       "      <td>0.199</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>32.95N</td>\n",
       "      <td>96.70W</td>\n",
       "      <td>1950-02-01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>405836</th>\n",
       "      <td>1950-02-01</td>\n",
       "      <td>1.655</td>\n",
       "      <td>0.057</td>\n",
       "      <td>Arlington</td>\n",
       "      <td>United States</td>\n",
       "      <td>39.38N</td>\n",
       "      <td>76.99W</td>\n",
       "      <td>1950-02-01</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                dt  AverageTemperature  AverageTemperatureUncertainty  \\\n",
       "402597  1950-02-01              11.144                          0.199   \n",
       "405836  1950-02-01               1.655                          0.057   \n",
       "\n",
       "             City        Country Latitude Longitude convertedDate  \n",
       "402597  Arlington  United States   32.95N    96.70W    1950-02-01  \n",
       "405836  Arlington  United States   39.38N    76.99W    1950-02-01  "
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_temperature[(df_temperature['City'] == 'Arlington') & (df_temperature.dt == '1950-02-01')]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "It looks like the temperature is measured in multiple locations for each city.\n",
    "When creating the dimension table, we'll compute the average temperatures and uncertainties per city"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Airport data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Next, we look at the contents of the airport dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(55075, 12)"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "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>ident</th>\n",
       "      <th>type</th>\n",
       "      <th>name</th>\n",
       "      <th>elevation_ft</th>\n",
       "      <th>continent</th>\n",
       "      <th>iso_country</th>\n",
       "      <th>iso_region</th>\n",
       "      <th>municipality</th>\n",
       "      <th>gps_code</th>\n",
       "      <th>iata_code</th>\n",
       "      <th>local_code</th>\n",
       "      <th>coordinates</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>00A</td>\n",
       "      <td>heliport</td>\n",
       "      <td>Total Rf Heliport</td>\n",
       "      <td>11.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-PA</td>\n",
       "      <td>Bensalem</td>\n",
       "      <td>00A</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00A</td>\n",
       "      <td>-74.93360137939453, 40.07080078125</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>00AA</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Aero B Ranch Airport</td>\n",
       "      <td>3435.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-KS</td>\n",
       "      <td>Leoti</td>\n",
       "      <td>00AA</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00AA</td>\n",
       "      <td>-101.473911, 38.704022</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>00AK</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Lowell Field</td>\n",
       "      <td>450.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-AK</td>\n",
       "      <td>Anchor Point</td>\n",
       "      <td>00AK</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00AK</td>\n",
       "      <td>-151.695999146, 59.94919968</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>00AL</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Epps Airpark</td>\n",
       "      <td>820.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-AL</td>\n",
       "      <td>Harvest</td>\n",
       "      <td>00AL</td>\n",
       "      <td>NaN</td>\n",
       "      <td>00AL</td>\n",
       "      <td>-86.77030181884766, 34.86479949951172</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>00AR</td>\n",
       "      <td>closed</td>\n",
       "      <td>Newport Hospital &amp; Clinic Heliport</td>\n",
       "      <td>237.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-AR</td>\n",
       "      <td>Newport</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-91.254898, 35.6087</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  ident           type                                name  elevation_ft  \\\n",
       "0   00A       heliport                   Total Rf Heliport          11.0   \n",
       "1  00AA  small_airport                Aero B Ranch Airport        3435.0   \n",
       "2  00AK  small_airport                        Lowell Field         450.0   \n",
       "3  00AL  small_airport                        Epps Airpark         820.0   \n",
       "4  00AR         closed  Newport Hospital & Clinic Heliport         237.0   \n",
       "\n",
       "  continent iso_country iso_region  municipality gps_code iata_code  \\\n",
       "0       NaN          US      US-PA      Bensalem      00A       NaN   \n",
       "1       NaN          US      US-KS         Leoti     00AA       NaN   \n",
       "2       NaN          US      US-AK  Anchor Point     00AK       NaN   \n",
       "3       NaN          US      US-AL       Harvest     00AL       NaN   \n",
       "4       NaN          US      US-AR       Newport      NaN       NaN   \n",
       "\n",
       "  local_code                            coordinates  \n",
       "0        00A     -74.93360137939453, 40.07080078125  \n",
       "1       00AA                 -101.473911, 38.704022  \n",
       "2       00AK            -151.695999146, 59.94919968  \n",
       "3       00AL  -86.77030181884766, 34.86479949951172  \n",
       "4        NaN                    -91.254898, 35.6087  "
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check the countries where these airports are located"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "iso_country\n",
       "AD        2\n",
       "AE       57\n",
       "AF       64\n",
       "AG        3\n",
       "AI        1\n",
       "AL       13\n",
       "AM       13\n",
       "AO      104\n",
       "AQ       27\n",
       "AR      848\n",
       "AS        4\n",
       "AT      145\n",
       "AU     1963\n",
       "AW        1\n",
       "AZ       35\n",
       "BA       15\n",
       "BB        6\n",
       "BD       16\n",
       "BE      146\n",
       "BF       51\n",
       "BG      134\n",
       "BH        4\n",
       "BI        7\n",
       "BJ       10\n",
       "BL        1\n",
       "BM        3\n",
       "BN        2\n",
       "BO      197\n",
       "BQ        3\n",
       "BR     4334\n",
       "      ...  \n",
       "TM       21\n",
       "TN       15\n",
       "TO        6\n",
       "TR      124\n",
       "TT        3\n",
       "TV        3\n",
       "TW       65\n",
       "TZ      207\n",
       "UA      191\n",
       "UG       38\n",
       "UM        6\n",
       "US    22757\n",
       "UY       54\n",
       "UZ      176\n",
       "VA        1\n",
       "VC        6\n",
       "VE      592\n",
       "VG        3\n",
       "VI        9\n",
       "VN       50\n",
       "VU       32\n",
       "WF        2\n",
       "WS        4\n",
       "XK        6\n",
       "YE       25\n",
       "YT        1\n",
       "ZA      489\n",
       "ZM      103\n",
       "ZW      138\n",
       "ZZ        7\n",
       "Name: iso_country, Length: 243, dtype: int64"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.groupby('iso_country')['iso_country'].count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "This dataset contains airport data for numerous countries. Our immigration dataset only contains entries into the US, thus via airports based in the united states. Therefore, we'll be reducing the size of this dataset\n",
    "\n",
    "Before we do that, let's make sure there is no missing data in the iso_county field."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(247, 12)"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports[df_airports['iso_country'].isna()].shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "continent\n",
       "AF    247\n",
       "Name: continent, dtype: int64"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Let's quickly check the missing country values to see if the continent data is filled out\n",
    "df_airports[df_airports['iso_country'].isna()].groupby('continent')['continent'].count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "All the missing country data is for airports based in Africa.\n",
    "\n",
    "Therefore, the dataset can safely be reduced"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Since all missing values are in africa, we simly remove them from the dataset\n",
    "df_airports = df_airports[df_airports['iso_country'].fillna('').str.upper().str.contains('US')].copy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The type column  contains several values. Let's look at them:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "type\n",
       "balloonport          18\n",
       "closed             1326\n",
       "heliport           6265\n",
       "large_airport       170\n",
       "medium_airport      692\n",
       "seaplane_base       566\n",
       "small_airport     13720\n",
       "Name: type, dtype: int64"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.groupby('type')['type'].count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We will assume the following here:\n",
    "* closed indicates the aireport is closed\n",
    "* No immigration data is collected from balloonports, seaplane bases or heliport since these means of transportation are used for recreational purposes or very short distances\n",
    "\n",
    "Therefore, we filter out all rows with these values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "excludedValues = ['closed', 'heliport', 'seaplane_base', 'balloonport']\n",
    "df_airports = df_airports[~df_airports['type'].str.strip().isin(excludedValues)].copy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's look for other missing values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ident               0\n",
       "type                0\n",
       "name                0\n",
       "elevation_ft       63\n",
       "continent       14582\n",
       "iso_country         0\n",
       "iso_region          0\n",
       "municipality       50\n",
       "gps_code          399\n",
       "iata_code       12717\n",
       "local_code        199\n",
       "coordinates         0\n",
       "dtype: int64"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#We check againvalues:\n",
    "df_airports.isnull().sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The ident code cannot be used to join airport data with the immigration set. Ineed, the airport linked to the ident, local or iata code columns are very different from the definitions found in the data dictionary. Therefore we must use the municipality names to join our datasets.\n",
    "\n",
    "From our previous validation, we know that we have 50 values missing from our dataset.\n",
    "Let's look at some of these missing values to see if we can get the municipality name through some other means"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "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>ident</th>\n",
       "      <th>type</th>\n",
       "      <th>name</th>\n",
       "      <th>elevation_ft</th>\n",
       "      <th>continent</th>\n",
       "      <th>iso_country</th>\n",
       "      <th>iso_region</th>\n",
       "      <th>municipality</th>\n",
       "      <th>gps_code</th>\n",
       "      <th>iata_code</th>\n",
       "      <th>local_code</th>\n",
       "      <th>coordinates</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>7653</th>\n",
       "      <td>6XA4</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Zadow Airstrip</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-TX</td>\n",
       "      <td>NaN</td>\n",
       "      <td>6XA4</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-95.954353809, 29.991738550900003</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7887</th>\n",
       "      <td>74xa</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Gun Barrel City Airpark</td>\n",
       "      <td>385.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-TX</td>\n",
       "      <td>NaN</td>\n",
       "      <td>74XA</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-96.1456650496, 32.3551499558</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8082</th>\n",
       "      <td>79ID</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Kooskia (Clear Creek Int) Airport</td>\n",
       "      <td>1800.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-ID</td>\n",
       "      <td>NaN</td>\n",
       "      <td>79ID</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-115.869691372, 46.0488642914</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8114</th>\n",
       "      <td>79WT</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Ellensburg (Rotor Ranch) Airport</td>\n",
       "      <td>1962.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-WA</td>\n",
       "      <td>NaN</td>\n",
       "      <td>79WT</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-120.589778423, 47.091426059499994</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9055</th>\n",
       "      <td>8FA4</td>\n",
       "      <td>small_airport</td>\n",
       "      <td>Samsula / Coe Field</td>\n",
       "      <td>40.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>US</td>\n",
       "      <td>US-FL</td>\n",
       "      <td>NaN</td>\n",
       "      <td>8FA4</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>-81.1328315735, 29.0102045831</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     ident           type                               name  elevation_ft  \\\n",
       "7653  6XA4  small_airport                     Zadow Airstrip           NaN   \n",
       "7887  74xa  small_airport            Gun Barrel City Airpark         385.0   \n",
       "8082  79ID  small_airport  Kooskia (Clear Creek Int) Airport        1800.0   \n",
       "8114  79WT  small_airport   Ellensburg (Rotor Ranch) Airport        1962.0   \n",
       "9055  8FA4  small_airport                Samsula / Coe Field          40.0   \n",
       "\n",
       "     continent iso_country iso_region municipality gps_code iata_code  \\\n",
       "7653       NaN          US      US-TX          NaN     6XA4       NaN   \n",
       "7887       NaN          US      US-TX          NaN     74XA       NaN   \n",
       "8082       NaN          US      US-ID          NaN     79ID       NaN   \n",
       "8114       NaN          US      US-WA          NaN     79WT       NaN   \n",
       "9055       NaN          US      US-FL          NaN     8FA4       NaN   \n",
       "\n",
       "     local_code                         coordinates  \n",
       "7653        NaN   -95.954353809, 29.991738550900003  \n",
       "7887        NaN       -96.1456650496, 32.3551499558  \n",
       "8082        NaN       -115.869691372, 46.0488642914  \n",
       "8114        NaN  -120.589778423, 47.091426059499994  \n",
       "9055        NaN       -81.1328315735, 29.0102045831  "
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# We also verify that the municipality field is available for all airports\n",
    "df_airports[df_airports.municipality.isna()].head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "None of these appear to be useable in a way that could be automated if we were building a pipeline. Therefore, we'll just remove them from our dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_airports = df_airports[~df_airports['municipality'].isna()].copy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We convert the municipality column to upper case in order to be able to join it with our other datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_airports.municipality = df_airports.municipality.str.upper()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "iso_region\n",
       "US-AK      586\n",
       "US-AL      197\n",
       "US-AR      291\n",
       "US-AZ      214\n",
       "US-CA      551\n",
       "US-CO      288\n",
       "US-CT       56\n",
       "US-DC        2\n",
       "US-DE       36\n",
       "US-FL      522\n",
       "US-GA      365\n",
       "US-HI       35\n",
       "US-IA      232\n",
       "US-ID      238\n",
       "US-IL      579\n",
       "US-IN      486\n",
       "US-KS      372\n",
       "US-KY      164\n",
       "US-LA      281\n",
       "US-MA       79\n",
       "US-MD      157\n",
       "US-ME      122\n",
       "US-MI      379\n",
       "US-MN      361\n",
       "US-MO      411\n",
       "US-MS      211\n",
       "US-MT      255\n",
       "US-NC      349\n",
       "US-ND      297\n",
       "US-NE      259\n",
       "US-NH       54\n",
       "US-NJ      116\n",
       "US-NM      149\n",
       "US-NV      113\n",
       "US-NY      402\n",
       "US-OH      492\n",
       "US-OK      372\n",
       "US-OR      357\n",
       "US-PA      486\n",
       "US-RI       10\n",
       "US-SC      173\n",
       "US-SD      162\n",
       "US-TN      228\n",
       "US-TX     1546\n",
       "US-U-A       3\n",
       "US-UT      103\n",
       "US-VA      311\n",
       "US-VT       66\n",
       "US-WA      379\n",
       "US-WI      457\n",
       "US-WV       83\n",
       "US-WY       95\n",
       "Name: iso_region, dtype: int64"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airports.groupby('iso_region')['iso_region'].count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Looking at the state data, U-A seems like an error. State is used in combination with city name to join with city demographics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# apply len to the iso_region field to see which ones are longer than 5 characters since the field is a combination of US and state code\n",
    "df_airports['len'] = df_airports[\"iso_region\"].apply(len)\n",
    "# let's remove the codes that are incorrect.\n",
    "df_airports = df_airports[df_airports['len']==5].copy()\n",
    "# finally, let's extract the state code\n",
    "df_airports['state'] = df_airports['iso_region'].str.strip().str.split(\"-\", n = 1, expand = True)[1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Demographic data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's look at the demographic dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(2891, 12)"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's first convert the city to upper case and remove any leading and trailing spaces"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_demographics.City = df_demographics.City.str.upper().str.strip()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's look at missing values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "City                       0\n",
       "State                      0\n",
       "Median Age                 0\n",
       "Male Population            3\n",
       "Female Population          3\n",
       "Total Population           0\n",
       "Number of Veterans        13\n",
       "Foreign-born              13\n",
       "Average Household Size    16\n",
       "State Code                 0\n",
       "Race                       0\n",
       "Count                      0\n",
       "dtype: int64"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics.isnull().sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "This dataset looks relatively clean with few missing values of significance."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We will not try to fix any of the missing data for now.\n",
    "We'll fix any issues with these missing rows when loading our dimension tables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# remove any leading or trailing spaces and convert to upper case\n",
    "df_demographics.City = df_demographics.City.str.strip().str.upper()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check whether city and race would work as a primary key for this table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "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>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",
       "      <th>Race</th>\n",
       "      <th>Count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>177</th>\n",
       "      <td>WILMINGTON</td>\n",
       "      <td>Delaware</td>\n",
       "      <td>36.4</td>\n",
       "      <td>32680.0</td>\n",
       "      <td>39277.0</td>\n",
       "      <td>71957</td>\n",
       "      <td>3063.0</td>\n",
       "      <td>3336.0</td>\n",
       "      <td>2.45</td>\n",
       "      <td>DE</td>\n",
       "      <td>Asian</td>\n",
       "      <td>1193</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>210</th>\n",
       "      <td>LAKEWOOD</td>\n",
       "      <td>California</td>\n",
       "      <td>39.9</td>\n",
       "      <td>41523.0</td>\n",
       "      <td>40069.0</td>\n",
       "      <td>81592</td>\n",
       "      <td>4094.0</td>\n",
       "      <td>18274.0</td>\n",
       "      <td>3.13</td>\n",
       "      <td>CA</td>\n",
       "      <td>Hispanic or Latino</td>\n",
       "      <td>24987</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>238</th>\n",
       "      <td>GLENDALE</td>\n",
       "      <td>California</td>\n",
       "      <td>42.1</td>\n",
       "      <td>98181.0</td>\n",
       "      <td>102844.0</td>\n",
       "      <td>201025</td>\n",
       "      <td>4448.0</td>\n",
       "      <td>111510.0</td>\n",
       "      <td>2.69</td>\n",
       "      <td>CA</td>\n",
       "      <td>White</td>\n",
       "      <td>146718</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>300</th>\n",
       "      <td>SPRINGFIELD</td>\n",
       "      <td>Massachusetts</td>\n",
       "      <td>31.8</td>\n",
       "      <td>74744.0</td>\n",
       "      <td>79592.0</td>\n",
       "      <td>154336</td>\n",
       "      <td>5723.0</td>\n",
       "      <td>16226.0</td>\n",
       "      <td>2.81</td>\n",
       "      <td>MA</td>\n",
       "      <td>Asian</td>\n",
       "      <td>5606</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>549</th>\n",
       "      <td>BLOOMINGTON</td>\n",
       "      <td>Indiana</td>\n",
       "      <td>23.5</td>\n",
       "      <td>40588.0</td>\n",
       "      <td>43227.0</td>\n",
       "      <td>83815</td>\n",
       "      <td>2368.0</td>\n",
       "      <td>10033.0</td>\n",
       "      <td>2.33</td>\n",
       "      <td>IN</td>\n",
       "      <td>Asian</td>\n",
       "      <td>9801</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            City          State  Median Age  Male Population  \\\n",
       "177   WILMINGTON       Delaware        36.4          32680.0   \n",
       "210     LAKEWOOD     California        39.9          41523.0   \n",
       "238     GLENDALE     California        42.1          98181.0   \n",
       "300  SPRINGFIELD  Massachusetts        31.8          74744.0   \n",
       "549  BLOOMINGTON        Indiana        23.5          40588.0   \n",
       "\n",
       "     Female Population  Total Population  Number of Veterans  Foreign-born  \\\n",
       "177            39277.0             71957              3063.0        3336.0   \n",
       "210            40069.0             81592              4094.0       18274.0   \n",
       "238           102844.0            201025              4448.0      111510.0   \n",
       "300            79592.0            154336              5723.0       16226.0   \n",
       "549            43227.0             83815              2368.0       10033.0   \n",
       "\n",
       "     Average Household Size State Code                Race   Count  \n",
       "177                    2.45         DE               Asian    1193  \n",
       "210                    3.13         CA  Hispanic or Latino   24987  \n",
       "238                    2.69         CA               White  146718  \n",
       "300                    2.81         MA               Asian    5606  \n",
       "549                    2.33         IN               Asian    9801  "
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#primary key will be the combination of city name and race\n",
    "df_demographics[df_demographics[['City','Race']].duplicated()].head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Clearly, the combination of city and race is not sufficient to work as a primary key. Let's look at a specific example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "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>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",
       "      <th>Race</th>\n",
       "      <th>Count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>102</th>\n",
       "      <td>WILMINGTON</td>\n",
       "      <td>North Carolina</td>\n",
       "      <td>35.5</td>\n",
       "      <td>52346.0</td>\n",
       "      <td>63601.0</td>\n",
       "      <td>115947</td>\n",
       "      <td>5908.0</td>\n",
       "      <td>7401.0</td>\n",
       "      <td>2.24</td>\n",
       "      <td>NC</td>\n",
       "      <td>Asian</td>\n",
       "      <td>3152</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>177</th>\n",
       "      <td>WILMINGTON</td>\n",
       "      <td>Delaware</td>\n",
       "      <td>36.4</td>\n",
       "      <td>32680.0</td>\n",
       "      <td>39277.0</td>\n",
       "      <td>71957</td>\n",
       "      <td>3063.0</td>\n",
       "      <td>3336.0</td>\n",
       "      <td>2.45</td>\n",
       "      <td>DE</td>\n",
       "      <td>Asian</td>\n",
       "      <td>1193</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           City           State  Median Age  Male Population  \\\n",
       "102  WILMINGTON  North Carolina        35.5          52346.0   \n",
       "177  WILMINGTON        Delaware        36.4          32680.0   \n",
       "\n",
       "     Female Population  Total Population  Number of Veterans  Foreign-born  \\\n",
       "102            63601.0            115947              5908.0        7401.0   \n",
       "177            39277.0             71957              3063.0        3336.0   \n",
       "\n",
       "     Average Household Size State Code   Race  Count  \n",
       "102                    2.24         NC  Asian   3152  \n",
       "177                    2.45         DE  Asian   1193  "
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics[(df_demographics.City == 'WILMINGTON') & (df_demographics.Race == 'Asian')]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The differene between these two rows is the state. Let's add it in the primary key combination"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "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>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",
       "      <th>Race</th>\n",
       "      <th>Count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "Empty DataFrame\n",
       "Columns: [City, State, Median Age, Male Population, Female Population, Total Population, Number of Veterans, Foreign-born, Average Household Size, State Code, Race, Count]\n",
       "Index: []"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics[df_demographics[['City', 'State','Race']].duplicated()]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "There are no duplicate rows when we use this combination. We will use it as our primary key."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Immigration data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The data dictionary provided already contains a lot of details on the missing data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+--------------+-----+--------+\n",
      "|    cicid| i94yr|i94mon|i94cit|i94res|i94port|arrdate|i94mode|i94addr|depdate|i94bir|i94visa|count|dtadfile|visapost|occup|entdepa|entdepd|entdepu|matflag|biryear| dtaddto|gender|insnum|airline|        admnum|fltno|visatype|\n",
      "+---------+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+--------------+-----+--------+\n",
      "|5748517.0|2016.0|   4.0| 245.0| 438.0|    LOS|20574.0|    1.0|     CA|20582.0|  40.0|    1.0|  1.0|20160430|     SYD| null|      G|      O|   null|      M| 1976.0|10292016|     F|  null|     QF|9.495387003E10|00011|      B1|\n",
      "|5748518.0|2016.0|   4.0| 245.0| 438.0|    LOS|20574.0|    1.0|     NV|20591.0|  32.0|    1.0|  1.0|20160430|     SYD| null|      G|      O|   null|      M| 1984.0|10292016|     F|  null|     VA|9.495562283E10|00007|      B1|\n",
      "|5748519.0|2016.0|   4.0| 245.0| 438.0|    LOS|20574.0|    1.0|     WA|20582.0|  29.0|    1.0|  1.0|20160430|     SYD| null|      G|      O|   null|      M| 1987.0|10292016|     M|  null|     DL|9.495640653E10|00040|      B1|\n",
      "|5748520.0|2016.0|   4.0| 245.0| 438.0|    LOS|20574.0|    1.0|     WA|20588.0|  29.0|    1.0|  1.0|20160430|     SYD| null|      G|      O|   null|      M| 1987.0|10292016|     F|  null|     DL|9.495645143E10|00040|      B1|\n",
      "|5748521.0|2016.0|   4.0| 245.0| 438.0|    LOS|20574.0|    1.0|     WA|20588.0|  28.0|    1.0|  1.0|20160430|     SYD| null|      G|      O|   null|      M| 1988.0|10292016|     M|  null|     DL|9.495638813E10|00040|      B1|\n",
      "+---------+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+--------------+-----+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_immigration.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "First, let's check to see if cicid can be used as a primary key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# We create a view of the immigration dataset\n",
    "df_immigration.createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3096313"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_immigration.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+\n",
      "|count(DISTINCT cicid)|\n",
      "+---------------------+\n",
      "|              3096313|\n",
      "+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT COUNT (DISTINCT cicid)\n",
    "FROM immig_table\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We've been provided with a data dictionary for i94port where all the codes are 3 character long\n",
    "Let's check if the same applies to the codes in the dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|len|\n",
      "+---+\n",
      "|  3|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT LENGTH (i94port) AS len\n",
    "FROM immig_table\n",
    "GROUP BY len\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "No processing is needed to join this to our dictionary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Next, we need to do is to convert the arrdate field into something that can be used."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "All dates in SAS correspond to the number of days since 1960-01-01.\n",
    "Therfore, we compute the arrival dates by adding arrdate to 1960-01-01"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_immigration = spark.sql(\"SELECT *, date_add(to_date('1960-01-01'), arrdate) AS arrival_date FROM immig_table\")\n",
    "df_immigration.createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Next, we replace the data in the I94VISA columns\n",
    "The three categories are:\n",
    "*   1 = Business\n",
    "*   2 = Pleasure\n",
    "*   3 = Student"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"SELECT *, CASE \n",
    "                        WHEN i94visa = 1.0 THEN 'Business' \n",
    "                        WHEN i94visa = 2.0 THEN 'Pleasure'\n",
    "                        WHEN i94visa = 3.0 THEN 'Student'\n",
    "                        ELSE 'N/A' END AS visa_type \n",
    "                        \n",
    "                FROM immig_table\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"SELECT *, CASE \n",
    "                        WHEN depdate >= 1.0 THEN date_add(to_date('1960-01-01'), depdate)\n",
    "                        WHEN depdate IS NULL THEN NULL\n",
    "                        ELSE 'N/A' END AS departure_date \n",
    "                        \n",
    "                FROM immig_table\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|       0|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Let's check the results from our previous query to make sure there are no N/A values\n",
    "spark.sql(\"SELECT count(*) FROM immig_table WHERE departure_date = 'N/A'\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Now, let's make sure that departure_date > arrival_date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|     375|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM immig_table\n",
    "WHERE departure_date <= arrival_date\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We have 120 rows where the departure_date appears to be wrong. Let's see if we can make sense of the dates"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------------+\n",
      "|arrival_date|departure_date|\n",
      "+------------+--------------+\n",
      "|  2016-04-30|    2016-04-29|\n",
      "|  2016-04-30|    2016-04-28|\n",
      "|  2016-04-30|    2016-04-29|\n",
      "|  2016-04-05|    2012-04-14|\n",
      "|  2016-04-05|    2016-03-14|\n",
      "|  2016-04-14|    2016-03-03|\n",
      "|  2016-04-01|    2016-02-28|\n",
      "|  2016-04-04|    2016-03-07|\n",
      "|  2016-04-01|    2016-03-05|\n",
      "|  2016-04-05|    2016-03-07|\n",
      "+------------+--------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT arrival_date, departure_date\n",
    "FROM immig_table\n",
    "WHERE departure_date <= arrival_date\n",
    "\"\"\").show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "It's impossible to know how to fix these errors. Since the number of affected rows is relatively small, we'll simply drop the rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT *\n",
    "FROM immig_table\n",
    "WHERE departure_date >= arrival_date\n",
    "\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Lastly, let's check how many distinct values we get in the arrival and departure dates to see if we need to merge our two sets for the time dimension table we'll be using in our model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------+\n",
      "|count(DISTINCT departure_date)|\n",
      "+------------------------------+\n",
      "|                           174|\n",
      "+------------------------------+\n",
      "\n",
      "+----------------------------+\n",
      "|count(DISTINCT arrival_date)|\n",
      "+----------------------------+\n",
      "|                          30|\n",
      "+----------------------------+\n",
      "\n",
      "+------------------------------+\n",
      "|count(DISTINCT departure_date)|\n",
      "+------------------------------+\n",
      "|                            29|\n",
      "+------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#check distinct departure dates\n",
    "spark.sql(\"SELECT COUNT (DISTINCT departure_date) FROM immig_table \").show()\n",
    "#check distinct arrival dates\n",
    "spark.sql(\"SELECT COUNT (DISTINCT arrival_date) FROM immig_table \").show()\n",
    "#check the common values between the two sets\n",
    "spark.sql(\"\"\"   SELECT COUNT(DISTINCT departure_date) \n",
    "                FROM immig_table \n",
    "                WHERE departure_date IN (\n",
    "                    SELECT DISTINCT arrival_date FROM immig_table\n",
    "                ) \n",
    "                \"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since one value is missing, we will merge the two data sets to allow for our dim table to include both departure and arrival dates."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check the data for the various arrival modes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------+\n",
      "|i94mode|count(1)|\n",
      "+-------+--------+\n",
      "|   null|     238|\n",
      "|    1.0| 2871184|\n",
      "|    3.0|   61572|\n",
      "|    2.0|   17970|\n",
      "|    9.0|    2517|\n",
      "+-------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT i94mode, count(*)\n",
    "FROM immig_table\n",
    "GROUP BY i94mode\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The arrival modes definition as per the dictonary is:\n",
    "* 1 = 'Air'\n",
    "* 2 = 'Sea'\n",
    "* 3 = 'Land'\n",
    "* 9 = 'Not reported'\n",
    "We will only keep Air arrival since we're joining this with airport datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We will keep only arrival by air to ensure that our dataset can work with the airports dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check if there are any missing values in the age column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|      46|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM immig_table\n",
    "WHERE i94bir IS NULL\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since we have some missing values here, let's check the birthyear instead to see if it can be used"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|count(biryear)|\n",
      "+--------------+\n",
      "|             0|\n",
      "+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT COUNT(biryear) FROM immig_table WHERE biryear IS NULL\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check if the year of birth makes sense too"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+------------+\n",
      "|max(biryear)|min(biryear)|\n",
      "+------------+------------+\n",
      "|      2016.0|      1916.0|\n",
      "+------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT MAX(biryear), MIN(biryear) FROM immig_table WHERE biryear IS NOT NULL\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's look at the frequency of travellers who are at least 80 years old, ie born in 1936 or earlier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|   24694|\n",
      "+--------+\n",
      "\n",
      "+-------+--------+\n",
      "|biryear|count(1)|\n",
      "+-------+--------+\n",
      "| 1916.0|       8|\n",
      "| 1917.0|      16|\n",
      "| 1918.0|      21|\n",
      "| 1919.0|      36|\n",
      "| 1920.0|      34|\n",
      "| 1921.0|      69|\n",
      "| 1922.0|      89|\n",
      "| 1923.0|     155|\n",
      "| 1924.0|     209|\n",
      "| 1925.0|     274|\n",
      "| 1926.0|     414|\n",
      "| 1927.0|     569|\n",
      "| 1928.0|     792|\n",
      "| 1929.0|    1073|\n",
      "| 1930.0|    1442|\n",
      "| 1931.0|    1794|\n",
      "| 1932.0|    2239|\n",
      "| 1933.0|    2688|\n",
      "| 1934.0|    3442|\n",
      "| 1935.0|    4194|\n",
      "+-------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Number of travellers who are older than 80\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM immig_table \n",
    "WHERE biryear IS NOT NULL\n",
    "AND biryear <= 1936\n",
    "\"\"\").show()\n",
    "\n",
    "# frequency of travellers by birth year\n",
    "spark.sql(\"\"\"\n",
    "SELECT biryear, COUNT(*)\n",
    "FROM immig_table \n",
    "WHERE biryear IS NOT NULL\n",
    "AND biryear <= 1936\n",
    "GROUP BY biryear\n",
    "ORDER BY biryear ASC\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The age of the travellers who are 105 years old are outliera with only 8 observations (out of 3 millions) and olny 0.6% of the travellers are over the age of 80 which seems reasonable."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since the birth year is available for each row, we can compute the age. Let's check if computed values match the age"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+\n",
      "|difference|count(1)|\n",
      "+----------+--------+\n",
      "|       0.0| 2953435|\n",
      "+----------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT (2016-biryear)-i94bir AS difference, count(*) FROM immig_table WHERE i94bir IS NOT NULL GROUP BY difference\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "This technique yields the exact same results as the age when the field is available. We will use that instead of the age to fill in the missing values."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check the gender to see if the data is useable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+\n",
      "|gender|count(1)|\n",
      "+------+--------+\n",
      "|     F| 1228646|\n",
      "|  null|  407456|\n",
      "|     M| 1316305|\n",
      "|     U|     238|\n",
      "|     X|     836|\n",
      "+------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT gender, count(*) \n",
    "FROM immig_table\n",
    "GROUP BY gender\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since we'd like to retain the gender of the various travellers, we will filter out all the rows where the gender is missing or incorrect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"SELECT * FROM immig_table WHERE gender IN ('F', 'M')\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check the citizenship and residence data to see if any values are missing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|       0|\n",
      "+--------+\n",
      "\n",
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|       0|\n",
      "+--------+\n",
      "\n",
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|  114019|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#citizenship countries\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) \n",
    "FROM immig_table\n",
    "WHERE i94cit IS NULL\n",
    "\"\"\").show()\n",
    "\n",
    "#residence countries\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) \n",
    "FROM immig_table\n",
    "WHERE i94res IS NULL\n",
    "\"\"\").show()\n",
    "\n",
    "#reported address\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) \n",
    "FROM immig_table\n",
    "WHERE i94addr IS NULL\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The addresses (really the state of residence) are missing quite often. We won't use this field, relying instead on the port of entry as a proxy for the traveller's address"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|       0|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM immig_table\n",
    "WHERE visatype IS NULL\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Detailed vista type is available for all rows. \n",
    "Let's check the aggregation to make sure the visa types are unique to each category."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+--------+\n",
      "|visa_type|visatype|count(1)|\n",
      "+---------+--------+--------+\n",
      "| Business|      B1|  186610|\n",
      "| Business|      E1|    3182|\n",
      "| Business|      E2|   16227|\n",
      "| Business|     GMB|     132|\n",
      "| Business|       I|    2962|\n",
      "| Business|      I1|     214|\n",
      "| Business|      WB|  185857|\n",
      "| Pleasure|      B2|  967988|\n",
      "| Pleasure|      CP|   11785|\n",
      "| Pleasure|     CPL|       8|\n",
      "| Pleasure|     GMT|   79454|\n",
      "| Pleasure|     SBP|       2|\n",
      "| Pleasure|      WT| 1060229|\n",
      "|  Student|      F1|   27789|\n",
      "|  Student|      F2|    1774|\n",
      "|  Student|      M1|     708|\n",
      "|  Student|      M2|      30|\n",
      "+---------+--------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT visa_type, visatype, count(*)\n",
    "FROM immig_table\n",
    "GROUP BY visa_type, visatype\n",
    "ORDER BY visa_type, visatype\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The definitions for various detailed visa types are listed below. Some are unknown.\n",
    "We couldn't find definitions for all the visa types. We will retain the details since it might be of interest from a demographic standpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "* B1 visa is for business visits valid for up to a year\n",
    "* B2 visa is for pleasure visits valid for up to a year\n",
    "* CP could not find a definition\n",
    "* E2 investor visas allows foreign investors to enter and work inside of the United States based on a substantial investment\n",
    "* F1 visas are used by non-immigrant students for Academic and Language training Courses. \n",
    "* F2 visas are used by the dependents of F1 visa holders\n",
    "* GMT could not find a definition\n",
    "* M1 for students enrolled in non-academic or “vocational study”. Mechanical, language, cooking classes, etc...\n",
    "* WB Waiver Program (WT/WB Status) travel to the United States for tourism or business for stays of 90 days or less without obtaining a visa.\n",
    "* WT Waiver Program (WT/WB Status) travel to the United States for tourism or business for stays of 90 days or less without obtaining a visa."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since we have little information besides the detailed visa type and the aggregate visa type, we will simply keep the information in our dimension table. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's take a look at the occupation field."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------+\n",
      "|occup|      n|\n",
      "+-----+-------+\n",
      "| null|2538838|\n",
      "|  STU|   3275|\n",
      "|  OTH|    508|\n",
      "|  NRR|    299|\n",
      "|  MKT|    262|\n",
      "|  EXA|    175|\n",
      "|  ULS|    142|\n",
      "|  ADM|    119|\n",
      "|  GLS|    119|\n",
      "|  TIE|    108|\n",
      "|  MVC|     58|\n",
      "|  ENO|     55|\n",
      "|  CEO|     53|\n",
      "|  TIP|     49|\n",
      "|  LLJ|     45|\n",
      "|  RET|     44|\n",
      "|  CMP|     43|\n",
      "|  PHS|     42|\n",
      "|  UNP|     33|\n",
      "|  HMK|     30|\n",
      "+-----+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "SELECT occup, COUNT(*) AS n\n",
    "FROM immig_table\n",
    "GROUP BY occup\n",
    "ORDER BY n DESC, occup\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The field is missing most of the time and the values provided are abbreviations. We won't be using it in our data model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Several other fields are missing a lot of values or simply not used or documented and will be dropped."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We've completed our analysis of the tables. Let's build a conceptual model now"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_immigration = spark.sql(\"\"\"SELECT * FROM immig_table\"\"\") "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 3: Define the Data Model\n",
    "#### 3.1 Conceptual Data Model\n",
    "Since we're interested in the flow of travellers through the united states. The i94 data will serve as our fact table.\n",
    "Our **fact_immigration** table will be :\n",
    "* cicid,\n",
    "* citizenship_country,\n",
    "* residence_country,\n",
    "* city,\n",
    "* state,\n",
    "* arrival_date,\n",
    "* departure_date,\n",
    "* age,\n",
    "* visa_type,\n",
    "* detailed_visa_type,\n",
    "\n",
    "For our dimension tables, since our dataset only contains one month of data we will keep a record of the daily entries and provide the uses with four dimensions to aggregate our data:\\\n",
    "\n",
    "**dim_time** : to aggregate the data suing various time units: The fileds available will be:\n",
    "* date, \n",
    "* year, \n",
    "* month, \n",
    "* day, \n",
    "* week,\n",
    "* weekday,\n",
    "* dayofyear\n",
    "\n",
    "**dim_airports**: Used to determine the areas with the largest flow of travelers. Fileds included will be:\n",
    "* ident,\n",
    "* type, \n",
    "* name, \n",
    "* elevation_ft, \n",
    "* state,\n",
    "* municipality, \n",
    "* iata_code\n",
    "\n",
    "**dim_city_demographics**: To look at the demographic data of the areas with the most travelers and potentially look at the impact of the flow of travellers on the demographic data (if it were updated on a regular basis). The fiels available will be:\n",
    "* City, \n",
    "* state, \n",
    "* median_age, \n",
    "* male_population, \n",
    "* female_population, \n",
    "* total population\n",
    "* Foreign_born, \n",
    "* Average_Household_Size, \n",
    "* Race, \n",
    "* Count,\n",
    "\n",
    "**dim_temperatures**: to look at the temperature data of the cities where traveller entry and departure is being reported. The fields included will be: \n",
    "* date, \n",
    "* City,\n",
    "* average temperature, \n",
    "* average temperature uncertainty \n",
    "\n",
    "\n",
    "#### 3.2 Mapping Out Data Pipelines\n",
    "\n",
    "Many of data data cleaning steps were documented in adetailed fashion in the section 2. Here are the steps again:\n",
    "##### Data Extraction:\n",
    "* Load all the datasets from CSV and SAS data files;\n",
    "\n",
    "##### Data Transformation and Loading:\n",
    "\n",
    "##### fact_immigration:\n",
    "* Drop rows where the mode of arrival is not air travel\n",
    "* Drop rows with incorrect gender data\n",
    "* convert arrival and departure dates;\n",
    "* replace country codes with the character string equivalents\n",
    "* replace visa_type with character string\n",
    "* replace port of entry with city and state\n",
    "* filter out any row where the port of entry is not in the US\n",
    "* compute age in a new row using birth year and year of our current date.\n",
    "* insert data into our fact table\n",
    "* Write to parquet\n",
    "\n",
    "##### dim_temperature:\n",
    "* For the temperature table, drop all data for cities outside the united states;\n",
    "* For the temperature table, drop all data for dates before 1950 since airtravel wasn't possible before that date;\n",
    "* Convert city to upper case\n",
    "* Compute the average temperature and uncertainty over date+city partitions\n",
    "* Insert into the temperature table as is since our dataset since our dataset may include new cities in future dates;\n",
    "* Write to parquet\n",
    "\n",
    "##### dim_time:\n",
    "* Get all the arrival dates from the immigration data_set;\n",
    "* extract year, month, day, week from the date and insert all the values in the dim_time table;\n",
    "* Write to parquet\n",
    "\n",
    "##### dim_airports:\n",
    "* Remove all non us airports\n",
    "* Remove all invalid port of entries, ie: ['closed', 'heliport', 'seaplane_base', 'balloonport']\n",
    "* Remove all rows where municipalities are missing.\n",
    "* Convert municipality to upper case\n",
    "* Insert to our table\n",
    "* Write to parquet\n",
    "\n",
    "##### dim_city_demographics:\n",
    "* Convert to city names to upper case\n",
    "* Insert to our table\n",
    "* Write to parquet\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 4: Run Pipelines to Model the Data \n",
    "#### 4.1 Create the data model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Note that spark automatically reads all fields as strings in our CSV files whereas pandas usually correctly autodectects the data types (as seen below). Consquently, we'll read all the csv files using pandas dataframes and then convert them to spark dataframes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_demographics_spark = spark.read.format(\"csv\").option(\"header\", \"true\").option(\"delimiter\", \";\").load('us-cities-demographics.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- Median Age: string (nullable = true)\n",
      " |-- Male Population: string (nullable = true)\n",
      " |-- Female Population: string (nullable = true)\n",
      " |-- Total Population: string (nullable = true)\n",
      " |-- Number of Veterans: string (nullable = true)\n",
      " |-- Foreign-born: string (nullable = true)\n",
      " |-- Average Household Size: string (nullable = true)\n",
      " |-- State Code: string (nullable = true)\n",
      " |-- Race: string (nullable = true)\n",
      " |-- Count: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_demographics_spark.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "City                       object\n",
       "State                      object\n",
       "Median Age                float64\n",
       "Male Population           float64\n",
       "Female Population         float64\n",
       "Total Population            int64\n",
       "Number of Veterans        float64\n",
       "Foreign-born              float64\n",
       "Average Household Size    float64\n",
       "State Code                 object\n",
       "Race                       object\n",
       "Count                       int64\n",
       "dtype: object"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_demographics.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "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: double (nullable = true)\n",
      " |-- Female Population: double (nullable = true)\n",
      " |-- Total Population: long (nullable = true)\n",
      " |-- Number of Veterans: double (nullable = true)\n",
      " |-- Foreign-born: double (nullable = true)\n",
      " |-- Average Household Size: double (nullable = true)\n",
      " |-- State Code: string (nullable = true)\n",
      " |-- Race: string (nullable = true)\n",
      " |-- Count: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.createDataFrame(df_demographics).printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Staging the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# load dictionary data\n",
    "df_countryCodes = pd.read_csv('countries.csv')\n",
    "df_i94portCodes = pd.read_csv('i94portCodes.csv')\n",
    "\n",
    "# load the various csv files into pandas dataframes\n",
    "df_demographics = pd.read_csv('us-cities-demographics.csv', sep=';')\n",
    "df_temperature = pd.read_csv('../../data2/GlobalLandTemperaturesByCity.csv')\n",
    "\n",
    "# load the SAS data\n",
    "df_immigration=spark.read.parquet(\"sas_data\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Transforming the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# let's convert the data dictionaries to views in our spark context to perform SQL operations with them\n",
    "spark_df_countryCodes = spark.createDataFrame(df_countryCodes)\n",
    "spark_df_countryCodes .createOrReplaceTempView(\"countryCodes\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# remove all entries with null values as they are either un reported or outside the US\n",
    "df_i94portCodes = df_i94portCodes[~df_i94portCodes.state.isna()].copy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# We need to exclude values for airports outside of the US. \n",
    "nonUSstates = ['CANADA', 'Canada', 'NETHERLANDS', 'NETH ANTILLES', 'THAILAND', 'ETHIOPIA', 'PRC', 'BERMUDA', 'COLOMBIA', 'ARGENTINA', 'MEXICO', \n",
    "               'BRAZIL', 'URUGUAY', 'IRELAND', 'GABON', 'BAHAMAS', 'MX', 'CAYMAN ISLAND', 'SEOUL KOREA', 'JAPAN', 'ROMANIA', 'INDONESIA',\n",
    "               'SOUTH AFRICA', 'ENGLAND', 'KENYA', 'TURK & CAIMAN', 'PANAMA', 'NEW GUINEA', 'ECUADOR', 'ITALY', 'EL SALVADOR']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_i94portCodes = df_i94portCodes[~df_i94portCodes.state.isin(nonUSstates)].copy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark_df_i94portCodes = spark.createDataFrame(df_i94portCodes)\n",
    "spark_df_i94portCodes .createOrReplaceTempView(\"i94portCodes\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_immigration.createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Remove all entries into the united states that weren't via air travel\n",
    "spark.sql(\"\"\"\n",
    "SELECT *\n",
    "FROM immig_table\n",
    "WHERE i94mode = 1\n",
    "\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# drop rows where the gender values entered is undefined\n",
    "spark.sql(\"\"\"SELECT * FROM immig_table WHERE gender IN ('F', 'M')\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# convert the arrival dates into a useable value\n",
    "spark.sql(\"SELECT *, date_add(to_date('1960-01-01'), arrdate) AS arrival_date FROM immig_table\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# convert the departure dates into a useable value\n",
    "spark.sql(\"\"\"SELECT *, CASE \n",
    "                        WHEN depdate >= 1.0 THEN date_add(to_date('1960-01-01'), depdate)\n",
    "                        WHEN depdate IS NULL THEN NULL\n",
    "                        ELSE 'N/A' END AS departure_date \n",
    "                        \n",
    "                FROM immig_table\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# we use an inner join to drop invalid codes\n",
    "#country of citizenship\n",
    "spark.sql(\"\"\"\n",
    "SELECT im.*, cc.country AS citizenship_country\n",
    "FROM immig_table im\n",
    "INNER JOIN countryCodes cc\n",
    "ON im.i94cit = cc.code\n",
    "\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#country of residence\n",
    "spark.sql(\"\"\"\n",
    "SELECT im.*, cc.country AS residence_country\n",
    "FROM immig_table im\n",
    "INNER JOIN countryCodes cc\n",
    "ON im.i94res = cc.code\n",
    "\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Add visa character string aggregation\n",
    "spark.sql(\"\"\"SELECT *, CASE \n",
    "                        WHEN i94visa = 1.0 THEN 'Business' \n",
    "                        WHEN i94visa = 2.0 THEN 'Pleasure'\n",
    "                        WHEN i94visa = 3.0 THEN 'Student'\n",
    "                        ELSE 'N/A' END AS visa_type \n",
    "                        \n",
    "                FROM immig_table\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Add entry_port names and entry port states to the view\n",
    "spark.sql(\"\"\"\n",
    "SELECT im.*, pc.location AS entry_port, pc.state AS entry_port_state\n",
    "FROM immig_table im \n",
    "INNER JOIN i94portCodes pc\n",
    "ON im.i94port = pc.code\n",
    "\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Compute the age of each individual and add it to the view\n",
    "spark.sql(\"\"\"\n",
    "SELECT *, (2016-biryear) AS age \n",
    "FROM immig_table\n",
    "\"\"\").createOrReplaceTempView(\"immig_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Insert the immigration fact data into a spark dataframe\n",
    "fact_immigration = spark.sql(\"\"\"\n",
    "                        SELECT \n",
    "                            cicid, \n",
    "                            citizenship_country,\n",
    "                            residence_country,\n",
    "                            TRIM(UPPER (entry_port)) AS city,\n",
    "                            TRIM(UPPER (entry_port_state)) AS state,\n",
    "                            arrival_date,\n",
    "                            departure_date,\n",
    "                            age,\n",
    "                            visa_type,\n",
    "                            visatype AS detailed_visa_type\n",
    "\n",
    "                        FROM immig_table\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# extract all distinct dates from arrival and departure dates to create dimension table\n",
    "dim_time = spark.sql(\"\"\"\n",
    "SELECT DISTINCT arrival_date AS date\n",
    "FROM immig_table\n",
    "UNION\n",
    "SELECT DISTINCT departure_date AS date\n",
    "FROM immig_table\n",
    "WHERE departure_date IS NOT NULL\n",
    "\"\"\")\n",
    "dim_time.createOrReplaceTempView(\"dim_time_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# extract year, month, day, weekofyear, dayofweek and weekofyear from the date and insert all the values in the dim_time table;\n",
    "dim_time = spark.sql(\"\"\"\n",
    "SELECT date, YEAR(date) AS year, MONTH(date) AS month, DAY(date) AS day, WEEKOFYEAR(date) AS week, DAYOFWEEK(date) as weekday, DAYOFYEAR(date) year_day\n",
    "FROM dim_time_table\n",
    "ORDER BY date ASC\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Keep only data for the United States\n",
    "df_temperature = df_temperature[df_temperature['Country']=='United States'].copy()\n",
    "\n",
    "# Convert the date to datetime objects\n",
    "df_temperature['date'] = pd.to_datetime(df_temperature.dt)\n",
    "\n",
    "# Remove all dates prior to 1950\n",
    "df_temperature=df_temperature[df_temperature['date']>\"1950-01-01\"].copy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# convert the city names to upper case\n",
    "df_temperature.City = df_temperature.City.str.strip().str.upper() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# convert the dataframes from pandas to spark\n",
    "spark_df_temperature = spark.createDataFrame(df_temperature)\n",
    "spark_df_temperature .createOrReplaceTempView(\"temperature\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "dim_temperature = spark.sql(\"\"\"\n",
    "SELECT\n",
    "    DISTINCT date, city,\n",
    "    AVG(AverageTemperature) OVER (PARTITION BY date, City) AS average_temperature, \n",
    "    AVG(AverageTemperatureUncertainty)  OVER (PARTITION BY date, City) AS average_termperature_uncertainty\n",
    "    \n",
    "FROM temperature\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "df_demographics.City = df_demographics.City.str.strip().str.upper()\n",
    "df_demographics['State Code'] = df_demographics['State Code'].str.strip().str.upper()\n",
    "df_demographics.Race = df_demographics.Race.str.strip().str.upper()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# convert the dataframes from pandas to spark\n",
    "spark_df_demographics = spark.createDataFrame(df_demographics)\n",
    "spark_df_demographics.createOrReplaceTempView(\"demographics\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# insert data into the demographics dim table\n",
    "dim_demographics = spark.sql(\"\"\"\n",
    "                                SELECT  City, \n",
    "                                        State, \n",
    "                                        `Median Age` AS median_age, \n",
    "                                        `Male Population` AS male_population, \n",
    "                                        `Female Population` AS female_population, \n",
    "                                        `Total Population` AS total_population, \n",
    "                                        `Foreign-born` AS foreign_born, \n",
    "                                        `Average Household Size` AS average_household_size, \n",
    "                                        `State Code` AS state_code, \n",
    "                                        Race, \n",
    "                                        Count\n",
    "                                FROM demographics\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#The airport dataset contains a lot of nulls. We'll load the csv directly into a spark dataframe to avoid having to deal with converting pandas NaN into nulls\n",
    "spark_df_airports = spark.read.format(\"csv\").option(\"header\", \"true\").load('airport-codes_csv.csv')\n",
    "spark_df_airports.createOrReplaceTempView(\"airports\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#equivalent to the following pandas code:\n",
    "# df_airports = df_airports[df_airports['iso_country'].fillna('').str.upper().str.contains('US')].copy()\n",
    "spark.sql(\"\"\"\n",
    "SELECT *\n",
    "FROM airports\n",
    "WHERE iso_country IS NOT NULL\n",
    "AND UPPER(TRIM(iso_country)) LIKE 'US'\n",
    "\"\"\").createOrReplaceTempView(\"airports\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#equivalent to the following pandas code:\n",
    "# excludedValues = ['closed', 'heliport', 'seaplane_base', 'balloonport']\n",
    "# df_airports = df_airports[~df_airports['type'].str.strip().isin(excludedValues)].copy()\n",
    "# df_airports = df_airports[~df_airports['municipality'].isna()].copy()\n",
    "# df_airports = df_airports[~df_airports['municipality'].isna()].copy()\n",
    "# df_airports['len'] = df_airports[\"iso_region\"].apply(len)\n",
    "# df_airports = df_airports[df_airports['len']==5].copy()\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "SELECT *\n",
    "FROM airports\n",
    "WHERE LOWER(TRIM(type)) NOT IN ('closed', 'heliport', 'seaplane_base', 'balloonport')\n",
    "AND municipality IS NOT NULL\n",
    "AND LENGTH(iso_region) = 5\n",
    "\"\"\").createOrReplaceTempView(\"airports\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "dim_airports = spark.sql(\"\"\"\n",
    "SELECT TRIM(ident) AS ident, type, name, elevation_ft, SUBSTR(iso_region, 4) AS state, TRIM(UPPER(municipality)) AS municipality, iata_code\n",
    "FROM airports\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Saving the data in parquet format\n",
    "dim_demographics.write.parquet(\"dim_demographics\")\n",
    "dim_time.write.parquet(\"dim_time\")\n",
    "dim_airports.write.parquet(\"dim_airports\")\n",
    "dim_temperature.write.parquet(\"dim_temperature\")\n",
    "fact_immigration.write.parquet(\"fact_immigration\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### 4.2 Data Quality Checks\n",
    "Explain the data quality checks you'll perform to ensure the pipeline ran as expected. These could include:\n",
    " * Integrity constraints on the relational database (e.g., unique key, data type, etc.)\n",
    " * Unit tests for the scripts to ensure they are doing the right thing\n",
    " * Source/Count checks to ensure completeness\n",
    " \n",
    "Run Quality Checks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#Let's check some things in our data\n",
    "dim_demographics.createOrReplaceTempView(\"dim_demographics\")\n",
    "dim_time.createOrReplaceTempView(\"dim_time\")\n",
    "dim_airports.createOrReplaceTempView(\"dim_airports\")\n",
    "dim_temperature.createOrReplaceTempView(\"dim_temperature\")\n",
    "fact_immigration.createOrReplaceTempView(\"fact_immigration\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "First, let's make sure the columns used as primary keys don't contain any null values.\n",
    "We define a function that could be incorporated in an automated data pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# we define the following function to check for null values\n",
    "def nullValueCheck(spark_ctxt, tables_to_check):\n",
    "    \"\"\"\n",
    "    This function performs null value checks on specific columns of given tables received as parameters and raises a ValueError exception when null values are encountered.\n",
    "    It receives the following parameters:\n",
    "    spark_ctxt: spark context where the data quality check is to be performed\n",
    "    tables_to_check: A dictionary containing (table, columns) pairs specifying for each table, which column is to be checked for null values.   \n",
    "    \"\"\"  \n",
    "    for table in tables_to_check:\n",
    "        print(f\"Performing data quality check on table {table}...\")\n",
    "        for column in tables_to_check[table]:\n",
    "            returnedVal = spark_ctxt.sql(f\"\"\"SELECT COUNT(*) as nbr FROM {table} WHERE {column} IS NULL\"\"\")\n",
    "            if returnedVal.head()[0] > 0:\n",
    "                raise ValueError(f\"Data quality check failed! Found NULL values in {column} column!\")\n",
    "        print(f\"Table {table} passed.\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Next we run the data quality check on all the tables in our data model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Performing data quality check on table fact_immigration...\n",
      "Table fact_immigration passed.\n",
      "Performing data quality check on table dim_time...\n",
      "Table dim_time passed.\n",
      "Performing data quality check on table dim_demographics...\n",
      "Table dim_demographics passed.\n",
      "Performing data quality check on table dim_airports...\n",
      "Table dim_airports passed.\n",
      "Performing data quality check on table dim_temperature...\n",
      "Table dim_temperature passed.\n"
     ]
    }
   ],
   "source": [
    "#dictionary of tables and columns to be checked\n",
    "tables_to_check = { 'fact_immigration' : ['cicid'], 'dim_time':['date'], 'dim_demographics': ['City','state_code'], 'dim_airports':['ident'], 'dim_temperature':['date','City']}\n",
    "\n",
    "#We call our function on the spark context\n",
    "nullValueCheck(spark, tables_to_check)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "The data quality check was successful.\n",
    "\n",
    "Let's do a more detailed check for each table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+\n",
      "|(count(1) - CAST(192 AS BIGINT))|\n",
      "+--------------------------------+\n",
      "|                               0|\n",
      "+--------------------------------+\n",
      "\n",
      "+--------------------------------------------+\n",
      "|(count(DISTINCT date) - CAST(192 AS BIGINT))|\n",
      "+--------------------------------------------+\n",
      "|                                           0|\n",
      "+--------------------------------------------+\n",
      "\n",
      "+----+\n",
      "|date|\n",
      "+----+\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#time dimension verification\n",
    "\n",
    "#check the number of rows in our time table : 192 expected\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*) - 192\n",
    "FROM dim_time\n",
    "\"\"\").show()\n",
    "\n",
    "# make sure each row has a distinct date key : 192 expected\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(DISTINCT date) - 192\n",
    "FROM dim_time\n",
    "\"\"\").show()\n",
    "\n",
    "# we could also subtract the result of one query from the other\n",
    "\n",
    "\n",
    "# and make sure all dates from the fact table are included in the time dimension (NULL is the expected result)\n",
    "spark.sql(\"\"\"\n",
    "SELECT DISTINCT date\n",
    "FROM dim_time\n",
    "\n",
    "MINUS\n",
    "\n",
    "(SELECT DISTINCT arrival_date AS date\n",
    "FROM immig_table\n",
    "UNION\n",
    "SELECT DISTINCT departure_date AS date\n",
    "FROM immig_table\n",
    "WHERE departure_date IS NOT NULL)\n",
    "\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------------------+\n",
      "|(count(DISTINCT cicid) - CAST(2165257 AS BIGINT))|\n",
      "+-------------------------------------------------+\n",
      "|                                                0|\n",
      "+-------------------------------------------------+\n",
      "\n",
      "+-------------------------------------------------+\n",
      "|(count(DISTINCT cicid) - CAST(2165257 AS BIGINT))|\n",
      "+-------------------------------------------------+\n",
      "|                                                0|\n",
      "+-------------------------------------------------+\n",
      "\n",
      "+------------------------------------+\n",
      "|(count(1) - CAST(2165257 AS BIGINT))|\n",
      "+------------------------------------+\n",
      "|                                   0|\n",
      "+------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#immigration verification\n",
    "\n",
    "# The number of primary key from the staging table (2165257 expected)\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(distinct cicid) - 2165257\n",
    "FROM immig_table\n",
    "\"\"\").show()\n",
    "\n",
    "#should match the primary key count from the fact table (2165257 expected)\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(distinct cicid) - 2165257\n",
    "FROM fact_immigration\n",
    "\"\"\").show()\n",
    "\n",
    "#and should match the row count from the fact table since it is also the primary key (2165257 expected)\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) - 2165257\n",
    "FROM fact_immigration\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------------+\n",
      "|(count(1) - CAST(2891 AS BIGINT))|\n",
      "+---------------------------------+\n",
      "|                                0|\n",
      "+---------------------------------+\n",
      "\n",
      "+----------------------------------------------------------+\n",
      "|(count(DISTINCT city, state, race) - CAST(2891 AS BIGINT))|\n",
      "+----------------------------------------------------------+\n",
      "|                                                         0|\n",
      "+----------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Let's check the demographics dimension table (2891 expected) \n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) - 2891\n",
    "FROM dim_demographics\n",
    "\"\"\").show()\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(DISTINCT city, state, race) - 2891\n",
    "FROM dim_demographics\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------+\n",
      "|(count(1) - CAST(14529 AS BIGINT))|\n",
      "+----------------------------------+\n",
      "|                                 0|\n",
      "+----------------------------------+\n",
      "\n",
      "+-----------------------------------------------+\n",
      "|(count(DISTINCT ident) - CAST(14529 AS BIGINT))|\n",
      "+-----------------------------------------------+\n",
      "|                                              0|\n",
      "+-----------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Let's check the primary key for airports (expected 14529)\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) - 14529\n",
    "FROM dim_airports\n",
    "\"\"\").show()\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(DISTINCT ident) - 14529\n",
    "FROM dim_airports\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------+\n",
      "|(count(1) - CAST(189472 AS BIGINT))|\n",
      "+-----------------------------------+\n",
      "|                                  0|\n",
      "+-----------------------------------+\n",
      "\n",
      "+-----------------------------------------------------+\n",
      "|(count(DISTINCT date, city) - CAST(189472 AS BIGINT))|\n",
      "+-----------------------------------------------------+\n",
      "|                                                    0|\n",
      "+-----------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#finally, city + date is our primary key for the temperature (expected 189472)\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "SELECT count(*) - 189472\n",
    "FROM dim_temperature\n",
    "\"\"\").show()\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(DISTINCT date, city) - 189472\n",
    "FROM dim_temperature\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Now, let's check what happens when we join our dimensions and fact tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-------------------+-----------------+------+-----+------------+--------------+----+---------+------------------+\n",
      "|    cicid|citizenship_country|residence_country|  city|state|arrival_date|departure_date| age|visa_type|detailed_visa_type|\n",
      "+---------+-------------------+-----------------+------+-----+------------+--------------+----+---------+------------------+\n",
      "|4041803.0|            GERMANY|          GERMANY|BANGOR|   ME|  2016-04-22|    2016-05-07|49.0| Business|                B1|\n",
      "|4041804.0|            GERMANY|          GERMANY|BANGOR|   ME|  2016-04-22|          null|38.0| Business|                B1|\n",
      "+---------+-------------------+-----------------+------+-----+------------+--------------+----+---------+------------------+\n",
      "only showing top 2 rows\n",
      "\n",
      "+-----+-------------+--------------------+------------+-----+------------+---------+\n",
      "|ident|         type|                name|elevation_ft|state|municipality|iata_code|\n",
      "+-----+-------------+--------------------+------------+-----+------------+---------+\n",
      "| 00AA|small_airport|Aero B Ranch Airport|        3435|   KS|       LEOTI|     null|\n",
      "| 00AK|small_airport|        Lowell Field|         450|   AK|ANCHOR POINT|     null|\n",
      "+-----+-------------+--------------------+------------+-----+------------+---------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# First, we join airport and immigration\n",
    "fact_immigration.show(2)\n",
    "dim_airports.show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Since a given city can have more than one airport and airport data is not provided in the immigration dataset, let's try to see how many city & state combinations are common to the two datasets.\n",
    "\n",
    "We're looking at immigrant influx based on cities. Thus, we'd like to check whether the use of city and state combination works well to match the data between dim_airport and fact_immigration "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------+\n",
      "|count(DISTINCT city, state)|\n",
      "+---------------------------+\n",
      "|                        151|\n",
      "+---------------------------+\n",
      "\n",
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|     102|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#here are the distinct combinations of city and state in our fact table\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(DISTINCT city, state)\n",
    "FROM fact_immigration\n",
    "\"\"\").show()\n",
    "\n",
    "# and the combinations of city and state that are common to both\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM\n",
    "(\n",
    "SELECT DISTINCT city, state\n",
    "FROM fact_immigration\n",
    ") fi\n",
    "INNER JOIN \n",
    "(\n",
    "SELECT DISTINCT municipality, state\n",
    "FROM dim_airports \n",
    ") da\n",
    "ON fi.city = da.municipality\n",
    "AND fi.state = da.state\n",
    "\"\"\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Roughly two thirds of our data in the fact table can be paired with data in the airport fact table. Considering that the immigration table only includes one month of data, this is quite good. We would normally use a left join"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Let's check the same thing with the demographics table.\n",
    "We expect the results of the join to be lower since the table doesn't include all cities in the united states."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-------------------+-----------------+------+-----+------------+--------------+----+---------+------------------+\n",
      "|    cicid|citizenship_country|residence_country|  city|state|arrival_date|departure_date| age|visa_type|detailed_visa_type|\n",
      "+---------+-------------------+-----------------+------+-----+------------+--------------+----+---------+------------------+\n",
      "|4041803.0|            GERMANY|          GERMANY|BANGOR|   ME|  2016-04-22|    2016-05-07|49.0| Business|                B1|\n",
      "|4041804.0|            GERMANY|          GERMANY|BANGOR|   ME|  2016-04-22|          null|38.0| Business|                B1|\n",
      "+---------+-------------------+-----------------+------+-----+------------+--------------+----+---------+------------------+\n",
      "only showing top 2 rows\n",
      "\n",
      "+-------------+-------------+----------+---------------+-----------------+----------------+------------+----------------------+----------+------------------+-----+\n",
      "|         City|        State|median_age|male_population|female_population|total_population|foreign_born|average_household_size|state_code|              Race|Count|\n",
      "+-------------+-------------+----------+---------------+-----------------+----------------+------------+----------------------+----------+------------------+-----+\n",
      "|SILVER SPRING|     Maryland|      33.8|        40601.0|          41862.0|           82463|     30908.0|                   2.6|        MD|HISPANIC OR LATINO|25924|\n",
      "|       QUINCY|Massachusetts|      41.0|        44129.0|          49500.0|           93629|     32935.0|                  2.39|        MA|             WHITE|58723|\n",
      "+-------------+-------------+----------+---------------+-----------------+----------------+------------+----------------------+----------+------------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fact_immigration.show(2)\n",
    "dim_demographics.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------+\n",
      "|count(DISTINCT city, state)|\n",
      "+---------------------------+\n",
      "|                        151|\n",
      "+---------------------------+\n",
      "\n",
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|      69|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#here are the distinct combinations of city and state in our fact table\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(DISTINCT city, state)\n",
    "FROM fact_immigration\n",
    "\"\"\").show()\n",
    "\n",
    "# and the combinations of city and state that are common to both the fact table and the demographics table\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM\n",
    "(\n",
    "SELECT DISTINCT city, state\n",
    "FROM fact_immigration\n",
    ") fi\n",
    "INNER JOIN \n",
    "(\n",
    "SELECT DISTINCT City, state_code\n",
    "FROM dim_demographics \n",
    ") da\n",
    "ON fi.city = da.City\n",
    "AND fi.state = da.state_code\n",
    "\"\"\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "A little less than half the cities are accounted for in our demographics database which isn't surprising but still quite good."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We have the option of filtering out non existent city/state combinations from the data using a query similar to the one below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "| 1983869|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# We use a count to see how many rows we would keep using this strategy\n",
    "spark.sql(\"\"\"\n",
    "SELECT COUNT(*)\n",
    "FROM fact_immigration\n",
    "WHERE CONCAT(city, state) IN (\n",
    "    SELECT CONCAT(fi.city, fi.state)\n",
    "    FROM\n",
    "    (\n",
    "        SELECT DISTINCT city, state\n",
    "        FROM fact_immigration\n",
    "    ) fi\n",
    "    INNER JOIN \n",
    "    (\n",
    "        SELECT DISTINCT municipality, state\n",
    "        FROM dim_airports \n",
    "    ) da\n",
    "    ON fi.city = da.municipality\n",
    "    AND fi.state = da.state\n",
    ")\n",
    "\"\"\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We drop from 2 165 257 rows to 1 983 869 rows which is still quite good.\n",
    "However, we are assuming that our datasets are incomplete, especially the demographic data since it only includes cities with populations larger than 65,000 inhabitants and prefer to minimize the amount of data that is being left out of our final result."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### 4.3 Data dictionary \n",
    "Create a data dictionary for your data model. For each field, provide a brief description of what the data is and where it came from. You can include the data dictionary in the notebook or in a separate file."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Step 5: Complete Project Write Up\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Consdiering the significant size of the immigration dataset (~ 3 million rows) for only a month, combined with the temperature, airport and demographic dataset, the most sensible technology choice for such an approach would be spark, especially if we were to process data over a longer period of time."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "We stated at the beginning of this project that we were interested in:\n",
    "* the effects of temperature aon the volume of travellers, \n",
    "* the seasonality of travel \n",
    "* the connection between the volume of travel and the number of entry ports (ie airports) \n",
    "* the connection between the volume of travel and the demographics of various cities\n",
    "\n",
    "None of these phenomenons require a rapid update of our data. A monthly or quarterly update would be sufficient for the needs of this study "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Alternate requirement scenarios:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "How would our approach change if the problem had the following requireements:\n",
    "* The data was increased by 100x: Our data would be stored in an Amazon S3 bucket (instead of storing it in the EMR cluster along with the staging tables) and loaded to our staging tables. We would still use spark as it as our data processing platform since it is the best suited platform for very large datasets.\n",
    "* The data populates a dashboard that must be updated on a daily basis by 7am every day: We would use Apache Airflow to perform the ETL and data qualtiy validation.\n",
    "* The database needed to be accessed by 100+ people: Once the data is ready to be consumed, it would be stored in a postgres database on a redshift cluster that easily supports multiuser access. "
   ]
  },
  {
   "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
}