Capstone Project Template.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Project Title\n",
    "### Data Engineering Capstone Project\n",
    "\n",
    "#### Project Summary\n",
    "\n",
    "This project looks into using the immigration data to form patterns that can assist airlines in planning for flight capacity. The following fields will be fields of interest in this study:\n",
    "1. Origination Port\n",
    "2. Landing Port in US\n",
    "3. Nationality of visitor\n",
    "4. Class of admission\n",
    "5. Race percentage in Landing Port\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": {},
   "outputs": [],
   "source": [
    "# Do all imports and installs here\n",
    "import pandas as pd\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T\n",
    "from decimal import Decimal\n",
    "import datetime\n",
    "import os\n",
    "\n",
    "pd.set_option(\"display.max_colwidth\", 200)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://b11abd6a76c5:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.5</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>dend_capstone</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f49a4ab2210>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .master(\"local\") \\\n",
    "    .appName(\"dend_capstone\") \\\n",
    "    .config(\"spark.sql.repl.eagerEval.enabled\", True) \\\n",
    "    .config(\"spark.sql.repl.eagerEval.truncate\", 50) \\\n",
    "    .config(\"spark.jars.packages\", \"saurfang:spark-sas7bdat:2.1.0-s_2.11\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 1: Scope the Project and Gather Data\n",
    "\n",
    "#### Scope \n",
    "\n",
    "This project analyzes the immigration data and air travel patterns in consideration with the following parameters:\n",
    "1. Origination Port\n",
    "2. Landing Port in US\n",
    "3. Nationality of visitor\n",
    "4. Class of admission\n",
    "5. Race percentage in Landing Port\n",
    "\n",
    "#### Describe and Gather Data \n",
    "\n",
    "Datasets\n",
    "1. I94 Immigration Data: This data comes from the US National Tourism and Trade Office. A data dictionary is included in the workspace. [This](https://travel.trade.gov/research/reports/i94/historical/2016.html) is where the data comes from. There's a sample file so you can take a look at the data in csv format before reading it all in. You do not have to use the entire dataset, just use what you need to accomplish the goal you set at the beginning of the project.\n",
    "2. World Temperature Data: This dataset came from Kaggle. You can read more about it (GlobalLandTemperaturesByCity.csv) [here](https://www.kaggle.com/berkeleyearth/climate-change-earth-surface-temperature-data).\n",
    "3. U.S. City Demographic Data: This data comes from OpenSoft. You can read more about it [here](https://public.opendatasoft.com/explore/dataset/us-cities-demographics/export/).\n",
    "4. Non Immigrant Class of Admission - Description of class of admission of Non Immigrants. Data obtained from [here](https://www.dhs.gov/immigration-statistics/nonimmigrant/NonimmigrantCOA). This page is manually scrapped since the excel on Data.gov is not available.\n",
    "5. Port Code - Extracted from I94_SAS_Labels_Descriptions.SAS (ports.csv)\n",
    "6. Country Code - Extracted from I94_SAS_Labels_Descriptions.SAS (country_code.csv)\n",
    "7. Airlines Code - Airline code and information obtained from [here](https://openflights.org/data.html#airline)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>field</th>\n",
       "      <th>description</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>i94yr</td>\n",
       "      <td>4 digit year</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>i94mon</td>\n",
       "      <td>Numeric month</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>i94cit</td>\n",
       "      <td>3 character code representing city of origination</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>i94res</td>\n",
       "      <td>3 character code likely to be representing city of residence</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>i94port</td>\n",
       "      <td>3 character code representing port of arrival</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>arrdate</td>\n",
       "      <td>Date of arrival in USA as a SAS numeric field</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>i94mode</td>\n",
       "      <td>1 character code representing the method of transport to USA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>i94addr</td>\n",
       "      <td>2 character string representing state of stay in USA</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>depdate</td>\n",
       "      <td>Date of departure from USA as a SAS numeric field</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>i94bir</td>\n",
       "      <td>Age of Respondent in Years</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>i94visa</td>\n",
       "      <td>Visa codes collapsed into three categories 1 = Business 2 = Pleasure 3 = Student</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>count</td>\n",
       "      <td>Used for summary statistics</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>dtadfile</td>\n",
       "      <td>Character Date Field - Date added to I-94 Files</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>visapost</td>\n",
       "      <td>Department of State where where Visa was issued</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>14</th>\n",
       "      <td>occup</td>\n",
       "      <td>Occupation that will be performed in U.S.</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>15</th>\n",
       "      <td>entdepa</td>\n",
       "      <td>Arrival Flag - admitted or paroled into the U.S.</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>16</th>\n",
       "      <td>entdepd</td>\n",
       "      <td>Departure Flag - Departed</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>17</th>\n",
       "      <td>entdepu</td>\n",
       "      <td>Update Flag - Either apprehended</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>18</th>\n",
       "      <td>matflag</td>\n",
       "      <td>Match flag - Match of arrival and departure records</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>19</th>\n",
       "      <td>biryear</td>\n",
       "      <td>4 digit year of birth</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>20</th>\n",
       "      <td>dtaddto</td>\n",
       "      <td>Character Date Field - Date to which admitted to U.S. (allowed to stay until)</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>21</th>\n",
       "      <td>gender</td>\n",
       "      <td>Non-immigrant sex</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>22</th>\n",
       "      <td>insnum</td>\n",
       "      <td>INS number</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>23</th>\n",
       "      <td>airline</td>\n",
       "      <td>Airline used to arrive in U.S.</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>24</th>\n",
       "      <td>admnum</td>\n",
       "      <td>Admission Number</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>25</th>\n",
       "      <td>fltno</td>\n",
       "      <td>Flight number of Airline used to arrive in U.S.</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>26</th>\n",
       "      <td>visatype</td>\n",
       "      <td>Class of admission legally admitting the non-immigrant to temporarily stay in U.S.</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       field  \\\n",
       "0      i94yr   \n",
       "1     i94mon   \n",
       "2     i94cit   \n",
       "3     i94res   \n",
       "4    i94port   \n",
       "5    arrdate   \n",
       "6    i94mode   \n",
       "7    i94addr   \n",
       "8    depdate   \n",
       "9     i94bir   \n",
       "10   i94visa   \n",
       "11     count   \n",
       "12  dtadfile   \n",
       "13  visapost   \n",
       "14     occup   \n",
       "15   entdepa   \n",
       "16   entdepd   \n",
       "17   entdepu   \n",
       "18   matflag   \n",
       "19   biryear   \n",
       "20   dtaddto   \n",
       "21    gender   \n",
       "22    insnum   \n",
       "23   airline   \n",
       "24    admnum   \n",
       "25     fltno   \n",
       "26  visatype   \n",
       "\n",
       "                                                                           description  \n",
       "0                                                                         4 digit year  \n",
       "1                                                                        Numeric month  \n",
       "2                                    3 character code representing city of origination  \n",
       "3                         3 character code likely to be representing city of residence  \n",
       "4                                        3 character code representing port of arrival  \n",
       "5                                        Date of arrival in USA as a SAS numeric field  \n",
       "6                         1 character code representing the method of transport to USA  \n",
       "7                                 2 character string representing state of stay in USA  \n",
       "8                                    Date of departure from USA as a SAS numeric field  \n",
       "9                                                           Age of Respondent in Years  \n",
       "10   Visa codes collapsed into three categories 1 = Business 2 = Pleasure 3 = Student   \n",
       "11                                                         Used for summary statistics  \n",
       "12                                     Character Date Field - Date added to I-94 Files  \n",
       "13                                     Department of State where where Visa was issued  \n",
       "14                                           Occupation that will be performed in U.S.  \n",
       "15                                    Arrival Flag - admitted or paroled into the U.S.  \n",
       "16                                                           Departure Flag - Departed  \n",
       "17                                                    Update Flag - Either apprehended  \n",
       "18                                 Match flag - Match of arrival and departure records  \n",
       "19                                                               4 digit year of birth  \n",
       "20       Character Date Field - Date to which admitted to U.S. (allowed to stay until)  \n",
       "21                                                                   Non-immigrant sex  \n",
       "22                                                                          INS number  \n",
       "23                                                      Airline used to arrive in U.S.  \n",
       "24                                                                    Admission Number  \n",
       "25                                     Flight number of Airline used to arrive in U.S.  \n",
       "26  Class of admission legally admitting the non-immigrant to temporarily stay in U.S.  "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_schema = spark.read.csv(\"I94_SAS_Labels_Description.csv\", header=True).toPandas()\n",
    "df_schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Bad Data File\n",
    "\n",
    "During data exploration `i94_jun16_sub.sas7bdat` was found to have additional columns defined in the file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "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",
      " |-- validres: double (nullable = true)\n",
      " |-- delete_days: double (nullable = true)\n",
      " |-- delete_mexl: double (nullable = true)\n",
      " |-- delete_dup: double (nullable = true)\n",
      " |-- delete_visa: double (nullable = true)\n",
      " |-- delete_recdup: 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_i94bad = spark.read \\\n",
    "    .format(\"com.github.saurfang.sas.spark\") \\\n",
    "    .load(\"raw_sas_data/i94_jun16_sub.sas7bdat\")\n",
    "df_i94bad.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>cicid</th><th>i94yr</th><th>i94mon</th><th>i94cit</th><th>i94res</th><th>i94port</th><th>arrdate</th><th>i94mode</th><th>i94addr</th><th>depdate</th><th>i94bir</th><th>i94visa</th><th>count</th><th>validres</th><th>delete_days</th><th>delete_mexl</th><th>delete_dup</th><th>delete_visa</th><th>delete_recdup</th><th>dtadfile</th><th>visapost</th><th>occup</th><th>entdepa</th><th>entdepd</th><th>entdepu</th><th>matflag</th><th>biryear</th><th>dtaddto</th><th>gender</th><th>insnum</th><th>airline</th><th>admnum</th><th>fltno</th><th>visatype</th></tr>\n",
       "<tr><td>4.0</td><td>2016.0</td><td>6.0</td><td>135.0</td><td>135.0</td><td>XXX</td><td>20612.0</td><td>null</td><td>null</td><td>null</td><td>59.0</td><td>2.0</td><td>1.0</td><td>1.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>null</td><td>null</td><td>null</td><td>Z</td><td>null</td><td>U</td><td>null</td><td>1957.0</td><td>10032016</td><td>null</td><td>null</td><td>null</td><td>1.4938462027E10</td><td>null</td><td>WT</td></tr>\n",
       "<tr><td>5.0</td><td>2016.0</td><td>6.0</td><td>135.0</td><td>135.0</td><td>XXX</td><td>20612.0</td><td>null</td><td>null</td><td>null</td><td>50.0</td><td>2.0</td><td>1.0</td><td>1.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>null</td><td>null</td><td>null</td><td>Z</td><td>null</td><td>U</td><td>null</td><td>1966.0</td><td>10032016</td><td>null</td><td>null</td><td>null</td><td>1.7460063727E10</td><td>null</td><td>WT</td></tr>\n",
       "<tr><td>6.0</td><td>2016.0</td><td>6.0</td><td>213.0</td><td>213.0</td><td>XXX</td><td>20609.0</td><td>null</td><td>null</td><td>null</td><td>27.0</td><td>3.0</td><td>1.0</td><td>1.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>null</td><td>null</td><td>null</td><td>T</td><td>null</td><td>U</td><td>null</td><td>1989.0</td><td>D/S</td><td>null</td><td>null</td><td>null</td><td>1.679297785E9</td><td>null</td><td>F1</td></tr>\n",
       "<tr><td>7.0</td><td>2016.0</td><td>6.0</td><td>213.0</td><td>213.0</td><td>XXX</td><td>20611.0</td><td>null</td><td>null</td><td>null</td><td>23.0</td><td>3.0</td><td>1.0</td><td>1.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>null</td><td>null</td><td>null</td><td>T</td><td>null</td><td>U</td><td>null</td><td>1993.0</td><td>D/S</td><td>null</td><td>null</td><td>null</td><td>1.140963185E9</td><td>null</td><td>F1</td></tr>\n",
       "<tr><td>16.0</td><td>2016.0</td><td>6.0</td><td>245.0</td><td>245.0</td><td>XXX</td><td>20632.0</td><td>null</td><td>null</td><td>null</td><td>24.0</td><td>3.0</td><td>1.0</td><td>1.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>0.0</td><td>null</td><td>null</td><td>null</td><td>T</td><td>null</td><td>U</td><td>null</td><td>1992.0</td><td>D/S</td><td>null</td><td>null</td><td>null</td><td>1.934535285E9</td><td>null</td><td>F1</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-----+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+-----------+-----------+----------+-----------+-------------+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+---------------+-----+--------+\n",
       "|cicid| i94yr|i94mon|i94cit|i94res|i94port|arrdate|i94mode|i94addr|depdate|i94bir|i94visa|count|validres|delete_days|delete_mexl|delete_dup|delete_visa|delete_recdup|dtadfile|visapost|occup|entdepa|entdepd|entdepu|matflag|biryear| dtaddto|gender|insnum|airline|         admnum|fltno|visatype|\n",
       "+-----+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+-----------+-----------+----------+-----------+-------------+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+---------------+-----+--------+\n",
       "|  4.0|2016.0|   6.0| 135.0| 135.0|    XXX|20612.0|   null|   null|   null|  59.0|    2.0|  1.0|     1.0|        0.0|        0.0|       0.0|        0.0|          0.0|    null|    null| null|      Z|   null|      U|   null| 1957.0|10032016|  null|  null|   null|1.4938462027E10| null|      WT|\n",
       "|  5.0|2016.0|   6.0| 135.0| 135.0|    XXX|20612.0|   null|   null|   null|  50.0|    2.0|  1.0|     1.0|        0.0|        0.0|       0.0|        0.0|          0.0|    null|    null| null|      Z|   null|      U|   null| 1966.0|10032016|  null|  null|   null|1.7460063727E10| null|      WT|\n",
       "|  6.0|2016.0|   6.0| 213.0| 213.0|    XXX|20609.0|   null|   null|   null|  27.0|    3.0|  1.0|     1.0|        0.0|        0.0|       0.0|        0.0|          0.0|    null|    null| null|      T|   null|      U|   null| 1989.0|     D/S|  null|  null|   null|  1.679297785E9| null|      F1|\n",
       "|  7.0|2016.0|   6.0| 213.0| 213.0|    XXX|20611.0|   null|   null|   null|  23.0|    3.0|  1.0|     1.0|        0.0|        0.0|       0.0|        0.0|          0.0|    null|    null| null|      T|   null|      U|   null| 1993.0|     D/S|  null|  null|   null|  1.140963185E9| null|      F1|\n",
       "| 16.0|2016.0|   6.0| 245.0| 245.0|    XXX|20632.0|   null|   null|   null|  24.0|    3.0|  1.0|     1.0|        0.0|        0.0|       0.0|        0.0|          0.0|    null|    null| null|      T|   null|      U|   null| 1992.0|     D/S|  null|  null|   null|  1.934535285E9| null|      F1|\n",
       "+-----+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+-----------+-----------+----------+-----------+-------------+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+---------------+-----+--------+"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_i94bad.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Convert to parquet files for easier manipulation\n",
    "\n",
    "Original raw data has numeric types inferred incorrectly as doubles and bigger numbers like admnum gets converted into Scientific Notation. Supplying a schema does not appear to work as dtadfile has values that cannot be mapped correctly during the load. This part reads the raw data performs the correct casting and saves them as a parquet file. It also drops the columns that do not belong to the official schema in the files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_parquet_file(input_path, output_path, logger=(lambda message: None)):\n",
    "    schema = {\n",
    "        \"cicid\": T.IntegerType(),\n",
    "        \"i94yr\": T.IntegerType(),\n",
    "        \"i94mon\": T.IntegerType(),  \n",
    "        \"i94cit\": T.IntegerType(),  \n",
    "        \"i94res\": T.IntegerType(),  \n",
    "        \"i94port\": T.StringType(), \n",
    "        \"arrdate\": T.IntegerType(),\n",
    "        \"i94mode\": T.IntegerType(),\n",
    "        \"i94addr\": T.StringType(),\n",
    "        \"depdate\": T.IntegerType(),\n",
    "        \"i94bir\": T.IntegerType(),\n",
    "        \"i94visa\": T.IntegerType(),\n",
    "        \"count\": T.IntegerType(), \n",
    "        \"dtadfile\": T.StringType(),\n",
    "        \"visapost\": T.StringType(),\n",
    "        \"occup\": T.StringType(),\n",
    "        \"entdepa\": T.StringType(),\n",
    "        \"entdepd\": T.StringType(),\n",
    "        \"entdepu\": T.StringType(),\n",
    "        \"matflag\": T.StringType(),\n",
    "        \"biryear\": T.IntegerType(),\n",
    "        \"dtaddto\": T.StringType(), \n",
    "        \"gender\": T.StringType(),\n",
    "        \"insnum\": T.StringType(),\n",
    "        \"airline\": T.StringType(),\n",
    "        \"admnum\": T.LongType(),\n",
    "        \"fltno\": T.StringType(),\n",
    "        \"visatype\": T.StringType(),\n",
    "    }\n",
    "\n",
    "    df_target = spark.read \\\n",
    "        .format(\"com.github.saurfang.sas.spark\") \\\n",
    "        .load(input_path)\n",
    "    df_target.select([\n",
    "        F.col(key).cast(value).alias(key) for (key,value) in schema.items()\n",
    "    ]).write.mode('overwrite').parquet(output_path)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "for file in os.listdir(\"raw_sas_data/\"):\n",
    "    if not file.endswith(\"sas7bdat\"):\n",
    "        continue\n",
    "        \n",
    "    clean_parquet_file(f\"raw_sas_data/{file}\", f\"sas_data/{file}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Utility Functions and UDFs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "@F.udf(returnType=T.DateType())\n",
    "def sas_date_converter(sas_date):\n",
    "    \"\"\"\n",
    "    UDF that performs a SAS date numeric field to a Python date.\n",
    "    \n",
    "    :param sas_date: SAS date numeric field \n",
    "    \"\"\"    \n",
    "    if sas_date == None:\n",
    "        return None\n",
    "    return (datetime.datetime(1960, 1, 1) + datetime.timedelta(sas_date))\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "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": {},
   "source": [
    "##### Immigration Data of Interest\n",
    "\n",
    "###### Basic data filtering\n",
    "\n",
    "Filtering to keep only air travel and removing unused columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def immigration_etl(input_path=\"sas_data/*\", output_path=\"staging_data/immigration_etl_raw\", logger=(lambda message: None)):\n",
    "    \"\"\"\n",
    "    Function that performs extraction and transformation of the immigration data returning a raw dataframe\n",
    "    \n",
    "    :param input_path: path to immigration data in parquet format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"       \n",
    "    \n",
    "    logger(\"\\n### Starting immigration extraction and transformation ###\")\n",
    "    df = spark.read.parquet(input_path)\n",
    "    \n",
    "    logger(f\"\\n*** Start removing non air travel: {df.count()} records***\")\n",
    "    df = df.select(\n",
    "        F.col(\"admnum\").alias(\"admission_number\"), \n",
    "        F.col(\"i94cit\").alias(\"immigrant_city\"),\n",
    "        F.col(\"i94res\").alias(\"immigrant_residence\"),\n",
    "        F.col(\"i94port\").alias(\"arrival_port\"),\n",
    "        F.col(\"arrdate\"),\n",
    "        F.col(\"i94addr\").alias(\"us_residence\"),\n",
    "        F.col(\"depdate\"),\n",
    "        F.col(\"i94bir\").alias(\"age\"),\n",
    "        F.col(\"biryear\").alias(\"year_of_birth\"),\n",
    "        F.col(\"airline\"),\n",
    "        F.col(\"gender\")\n",
    "    ).where(df.i94mode == 1)\n",
    "    logger(f\"*** End removing non air travel: {df.count()} records***\")\n",
    "        \n",
    "    logger(f\"\\n*** Start dropping duplicates: {df.count()} records***\")\n",
    "    df = df.dropDuplicates()\n",
    "    logger(f\"*** Start dropping duplicates: {df.count()} records***\")\n",
    "                   \n",
    "    logger(f\"\\n*** Start null immigrant_city cleanup: {df.count()} records***\")\n",
    "    df = df.where(F.isnull(\"immigrant_city\") == False)\n",
    "    logger(f\"*** End null immigrant_city cleanup: {df.count()} records***\")  \n",
    "    logger(f\"\\n*** Start adding arrival_date_source ***\")\n",
    "    df = df.withColumn(\n",
    "        \"arrival_date_source\", sas_date_converter(F.col(\"arrdate\"))\n",
    "    )\n",
    "    logger(f\"*** End adding arrival_date_source ***\")\n",
    "\n",
    "    logger(f\"\\n*** Start adding departure_date_source ***\")\n",
    "    df = df.withColumn(\n",
    "        \"departure_date_source\", sas_date_converter(F.col(\"depdate\"))\n",
    "    )\n",
    "    logger(f\"*** End adding departure_date_source ***\")\n",
    "\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\") \n",
    "    \n",
    "    logger(\"\\n### End immigration extraction and transformation ###\")\n",
    "\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "### Starting immigration extraction and transformation ###\n",
      "\n",
      "*** Start removing non air travel: 3096313 records***\n",
      "*** End removing non air travel: 2994505 records***\n",
      "\n",
      "*** Start dropping duplicates: 2994505 records***\n",
      "*** Start dropping duplicates: 2994495 records***\n",
      "\n",
      "*** Start null immigrant_city cleanup: 2994495 records***\n",
      "*** End null immigrant_city cleanup: 2994495 records***\n",
      "\n",
      "*** Start adding arrival_date_source ***\n",
      "*** End adding arrival_date_source ***\n",
      "\n",
      "*** Start adding departure_date_source ***\n",
      "*** End adding departure_date_source ***\n",
      "\n",
      "*** Start writing 2994495 records to staging_data/immigration_etl_raw ***\n",
      "*** End writing 2994495 records to staging_data/immigration_etl_raw ***\n",
      "\n",
      "### End immigration extraction and transformation ###\n",
      "\n",
      "Checking for nulls\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>admission_number</th>\n",
       "      <th>immigrant_city</th>\n",
       "      <th>immigrant_residence</th>\n",
       "      <th>arrival_port</th>\n",
       "      <th>arrdate</th>\n",
       "      <th>us_residence</th>\n",
       "      <th>depdate</th>\n",
       "      <th>age</th>\n",
       "      <th>year_of_birth</th>\n",
       "      <th>airline</th>\n",
       "      <th>gender</th>\n",
       "      <th>arrival_date_source</th>\n",
       "      <th>departure_date_source</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>117410</td>\n",
       "      <td>123043</td>\n",
       "      <td>594</td>\n",
       "      <td>594</td>\n",
       "      <td>390</td>\n",
       "      <td>411902</td>\n",
       "      <td>0</td>\n",
       "      <td>123043</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   admission_number  immigrant_city  immigrant_residence  arrival_port  \\\n",
       "0                 0               0                    0             0   \n",
       "\n",
       "   arrdate  us_residence  depdate  age  year_of_birth  airline  gender  \\\n",
       "0        0        117410   123043  594            594      390  411902   \n",
       "\n",
       "   arrival_date_source  departure_date_source  \n",
       "0                    0                 123043  "
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "df_columns_of_interest = immigration_etl(\"sas_data/i94_apr16_sub.sas7bdat\", \"staging_data/immigration_etl_raw\", print)\n",
    "\n",
    "print(\"\\nChecking for nulls\")\n",
    "display(df_columns_of_interest.select([\n",
    "    F.count(F.when(F.isnull(c), c)).alias(c) for c in df_columns_of_interest.columns\n",
    "]).toPandas())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- admission_number: long (nullable = true)\n",
      " |-- immigrant_city: integer (nullable = true)\n",
      " |-- immigrant_residence: integer (nullable = true)\n",
      " |-- arrival_port: string (nullable = true)\n",
      " |-- arrdate: integer (nullable = true)\n",
      " |-- us_residence: string (nullable = true)\n",
      " |-- depdate: integer (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      " |-- year_of_birth: integer (nullable = true)\n",
      " |-- airline: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- arrival_date_source: date (nullable = true)\n",
      " |-- departure_date_source: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_columns_of_interest.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Airline Code\n",
    "\n",
    "###### Load and basic data filtering\n",
    "\n",
    "Airline code and information obtained from [here](https://openflights.org/data.html#airline)\n",
    "- Airline ID: Unique OpenFlights identifier for this airline.\n",
    "- Name: Name of the airline.\n",
    "- Alias: Alias of the airline. For example, All Nippon Airways is commonly known as \"ANA\".\n",
    "- IATA: 2-letter IATA code, if available.\n",
    "- ICAO: 3-letter ICAO code, if available.\n",
    "- Callsign: Airline callsign.\n",
    "- Country: Country or territory where airport is located. See Countries to cross-reference to ISO 3166-1 codes.\n",
    "- Active: \"Y\" if the airline is or has until recently been operational, \"N\" if it is defunct. This field is not reliable: in particular, major airlines that stopped flying long ago, but have not had their IATA code reassigned (eg. Ansett/AN), will incorrectly show as \"Y\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def airline_etl(input_path=\"airline.csv\", output_path=\"staging_data/airline_etl_raw\", logger=(lambda message: None)):\n",
    "    \"\"\"\n",
    "    Function that performs extraction and transformation of the airline data returning a raw dataframe\n",
    "    \n",
    "    :param input_path: path containing the airline data in csv format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"       \n",
    "    \n",
    "    logger(\"\\n### Starting airline extraction and transformation ###\")\n",
    "    df = spark.read.csv(\n",
    "        input_path, header=True\n",
    "    ).select(\n",
    "        F.col(\"IATA\").alias(\"iata\"),        \n",
    "        F.col(\"Name\").alias(\"name\"), \n",
    "        F.col(\"Alias\").alias(\"alias\"),\n",
    "        F.col(\"Callsign\").alias(\"callsign\"),\n",
    "        F.col(\"Country\").alias(\"country\")        \n",
    "    )\n",
    "    \n",
    "    logger(f\"\\n*** Start removing invalid IATA: {df.count()} records***\")\n",
    "    df=df.where((F.isnull(\"iata\") == False) & (F.length(\"iata\") == 2))\n",
    "    logger(f\"*** End removing invalid IATA: {df.count()} records***\")\n",
    "            \n",
    "    logger(f\"\\n*** Start dropping duplicates: {df.count()} records***\")\n",
    "    df = df.dropDuplicates([\"iata\"])\n",
    "    logger(f\"*** Start dropping duplicates: {df.count()} records***\")\n",
    "    \n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.select(\n",
    "        F.col(\"Name\").alias(\"name\"), \n",
    "        F.col(\"Alias\").alias(\"alias\"),\n",
    "        F.col(\"IATA\").alias(\"iata\"),\n",
    "        F.col(\"Callsign\").alias(\"callsign\"),\n",
    "        F.col(\"Country\").alias(\"country\")        \n",
    "    ).write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")    \n",
    "    \n",
    "    logger(\"\\n### End airline extraction and transformation ###\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "### Starting airline extraction and transformation ###\n",
      "\n",
      "*** Start removing invalid IATA: 6162 records***\n",
      "*** End removing invalid IATA: 1534 records***\n",
      "\n",
      "*** Start dropping duplicates: 1534 records***\n",
      "*** Start dropping duplicates: 1120 records***\n",
      "\n",
      "*** Start writing 1120 records to staging_data/airline_etl_raw ***\n",
      "*** End writing 1120 records to staging_data/airline_etl_raw ***\n",
      "\n",
      "### End airline extraction and transformation ###\n",
      "\n",
      "Checking for nulls\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>iata</th>\n",
       "      <th>name</th>\n",
       "      <th>alias</th>\n",
       "      <th>callsign</th>\n",
       "      <th>country</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1037</td>\n",
       "      <td>228</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   iata  name  alias  callsign  country\n",
       "0     0     0   1037       228        6"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>iata</th><th>name</th><th>alias</th><th>callsign</th><th>country</th></tr>\n",
       "<tr><td>07</td><td>Samurai Airlines</td><td>Samurai Airlines (DUMMY)</td><td>Sam</td><td>Pakistan</td></tr>\n",
       "<tr><td>1L</td><td>Open Skies Consultative Commission</td><td>null</td><td>OPEN SKIES</td><td>United States</td></tr>\n",
       "<tr><td>3P</td><td>Tiara Air</td><td>null</td><td>TIARA</td><td>Aruba</td></tr>\n",
       "<tr><td>DZ</td><td>Starline.kz</td><td>null</td><td>ALUNK</td><td>Kazakhstan</td></tr>\n",
       "<tr><td>LT</td><td>LTU International</td><td>null</td><td>LTU</td><td>Germany</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+----+----------------------------------+------------------------+----------+-------------+\n",
       "|iata|                              name|                   alias|  callsign|      country|\n",
       "+----+----------------------------------+------------------------+----------+-------------+\n",
       "|  07|                  Samurai Airlines|Samurai Airlines (DUMMY)|       Sam|     Pakistan|\n",
       "|  1L|Open Skies Consultative Commission|                    null|OPEN SKIES|United States|\n",
       "|  3P|                         Tiara Air|                    null|     TIARA|        Aruba|\n",
       "|  DZ|                       Starline.kz|                    null|     ALUNK|   Kazakhstan|\n",
       "|  LT|                 LTU International|                    null|       LTU|      Germany|\n",
       "+----+----------------------------------+------------------------+----------+-------------+"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_airline=airline_etl(\"airline.csv\", \"staging_data/airline_etl_raw\", print)\n",
    "print(\"\\nChecking for nulls\")\n",
    "display(df_airline.select([F.count(F.when(F.isnull(c), c)).alias(c) for c in df_airline.columns]).toPandas())\n",
    "df_airline.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- iata: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- alias: string (nullable = true)\n",
      " |-- callsign: string (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_airline.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Country Code\n",
    "\n",
    "###### Load and basic data filtering\n",
    "\n",
    "This is a datasource transform manually using data in I94_SAS_Labels_Descriptions.SAS since there is no reference datasource available. And codifying the transform does not produce enough benefit to justify the work involved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "def country_etl(input_path=\"country_code.csv\", output_path=\"staging_data/country_etl_raw\", logger=(lambda message: None)):\n",
    "    \"\"\"\n",
    "    Function that performs extraction and transformation of the country data returning a raw dataframe\n",
    "    \n",
    "    :param input_path: path containing the airline data in csv format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"       \n",
    "    schema = T.StructType([\n",
    "        T.StructField(\"code\", T.IntegerType(), False),\n",
    "        T.StructField(\"country\", T.StringType(), False)\n",
    "    ])\n",
    "    \n",
    "    logger(\"\\n### Starting country extraction and transformation ###\")\n",
    "    df = spark.read.csv(input_path, header=True, schema=schema)\n",
    "            \n",
    "    logger(f\"\\n*** Start dropping duplicates: {df.count()} records***\")\n",
    "    df = df.dropDuplicates([\"code\"])\n",
    "    logger(f\"*** End dropping duplicates: {df.count()} records***\")\n",
    "\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")     \n",
    "    \n",
    "    logger(\"\\n### End country extraction and transformation ###\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "### Starting country extraction and transformation ###\n",
      "\n",
      "*** Start dropping duplicates: 289 records***\n",
      "*** End dropping duplicates: 289 records***\n",
      "\n",
      "*** Start writing 289 records to staging_data/country_etl_raw ***\n",
      "*** End writing 289 records to staging_data/country_etl_raw ***\n",
      "\n",
      "### End country extraction and transformation ###\n",
      "\n",
      "Checking for nulls\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>code</th>\n",
       "      <th>country</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   code  country\n",
       "0     0        0"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>code</th><th>country</th></tr>\n",
       "<tr><td>471</td><td>INVALID: MARIANA ISLANDS, NORTHERN</td></tr>\n",
       "<tr><td>243</td><td>BURMA</td></tr>\n",
       "<tr><td>392</td><td>MALI</td></tr>\n",
       "<tr><td>737</td><td>INVALID: MIDWAY ISLANDS</td></tr>\n",
       "<tr><td>516</td><td>TRINIDAD AND TOBAGO</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+----+----------------------------------+\n",
       "|code|                           country|\n",
       "+----+----------------------------------+\n",
       "| 471|INVALID: MARIANA ISLANDS, NORTHERN|\n",
       "| 243|                             BURMA|\n",
       "| 392|                              MALI|\n",
       "| 737|           INVALID: MIDWAY ISLANDS|\n",
       "| 516|               TRINIDAD AND TOBAGO|\n",
       "+----+----------------------------------+"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_country_code = country_etl(\"country_code.csv\", \"staging_data/country_etl_raw\", print)\n",
    "print(\"\\nChecking for nulls\")\n",
    "display(df_country_code.select([F.count(F.when(F.isnull(c), c)).alias(c) for c in df_country_code.columns]).toPandas())\n",
    "df_country_code.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- code: integer (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_country_code.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Ports\n",
    "\n",
    "###### Load and basic data filtering\n",
    "\n",
    "This is a datasource transform manually using data in I94_SAS_Labels_Descriptions.SAS since there is no reference datasource available. And codifying the transform does not produce enough benefit to justify the work involved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "def port_etl(input_path=\"ports.csv\", output_path=\"staging_data/port_etl_raw\", logger=(lambda message: None)):\n",
    "    \"\"\"\n",
    "    Function that performs extraction and transformation of the ports data returning a raw dataframe\n",
    "    \n",
    "    :param input_path: path containing the ports data in csv format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"       \n",
    "    \n",
    "    logger(\"\\n### Starting ports extraction and transformation ###\")\n",
    "    df = spark.read.csv(input_path, header=True)\n",
    "            \n",
    "    logger(f\"\\n*** Start dropping duplicates: {df.count()} records***\")\n",
    "    df = df.dropDuplicates([\"code\"])\n",
    "    logger(f\"*** End dropping duplicates: {df.count()} records***\")\n",
    "\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")     \n",
    "    \n",
    "    logger(\"\\n### End ports extraction and transformation ###\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "### Starting ports extraction and transformation ###\n",
      "\n",
      "*** Start dropping duplicates: 660 records***\n",
      "*** End dropping duplicates: 660 records***\n",
      "\n",
      "*** Start writing 660 records to staging_data/port_etl_raw ***\n",
      "*** End writing 660 records to staging_data/port_etl_raw ***\n",
      "\n",
      "### End ports extraction and transformation ###\n",
      "\n",
      "Checking for nulls\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>code</th>\n",
       "      <th>country</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   code  country\n",
       "0     0        0"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>code</th><th>city</th><th>state</th><th>country</th></tr>\n",
       "<tr><td>BGM</td><td> BANGOR</td><td> ME            </td><td>US</td></tr>\n",
       "<tr><td>FMY</td><td> FORT MYERS</td><td> FL        </td><td>US</td></tr>\n",
       "<tr><td>LEB</td><td> LEBANON</td><td> NH           </td><td>US</td></tr>\n",
       "<tr><td>DNS</td><td> DUNSEITH</td><td> ND          </td><td>US</td></tr>\n",
       "<tr><td>EGL</td><td> EAGLE</td><td> AK             </td><td>US</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+----+-----------+----------------+-------+\n",
       "|code|       city|           state|country|\n",
       "+----+-----------+----------------+-------+\n",
       "| BGM|     BANGOR|  ME            |     US|\n",
       "| FMY| FORT MYERS|      FL        |     US|\n",
       "| LEB|    LEBANON|   NH           |     US|\n",
       "| DNS|   DUNSEITH|    ND          |     US|\n",
       "| EGL|      EAGLE| AK             |     US|\n",
       "+----+-----------+----------------+-------+"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_ports = port_etl(\"ports.csv\", \"staging_data/port_etl_raw\", print)\n",
    "print(\"\\nChecking for nulls\")\n",
    "display(df_country_code.select(\n",
    "    [F.count(F.when(F.isnull(c), c)).alias(c) for c in df_country_code.columns]\n",
    ").toPandas())\n",
    "df_ports.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- code: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_ports.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Non immigrant class of admission\n",
    "\n",
    "###### Load and basic data filtering\n",
    "\n",
    "Description of class of admission of Non Immigrants. Data obtained from [here](https://www.dhs.gov/immigration-statistics/nonimmigrant/NonimmigrantCOA). This page is manually scrapped since the excel on Data.gov is not available. Codifying the extract does not produce enough benefit to justify the work involved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "def class_of_admission_etl(input_path=\"non_immigrant_class_of_admission.csv\", output_path=\"staging_data/class_of_admission_etl_raw\", logger=(lambda message: None)):\n",
    "    \"\"\"\n",
    "    Function that performs extraction and transformation of the class of admission data returning a raw dataframe\n",
    "    \n",
    "    :param input_path: path containing the class of admission data in csv format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"       \n",
    "    \n",
    "    logger(\"\\n### Starting class of admission extraction and transformation ###\")\n",
    "    df = spark.read.csv(input_path, header=True)\n",
    "            \n",
    "    logger(f\"\\n*** Start dropping duplicates: {df.count()} records***\")\n",
    "    df = df.dropDuplicates([\"code\"])\n",
    "    logger(f\"*** End dropping duplicates: {df.count()} records***\")\n",
    "\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")     \n",
    "    \n",
    "    logger(\"\\n### End class of admission extraction and transformation ###\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "### Starting class of admission extraction and transformation ###\n",
      "\n",
      "*** Start dropping duplicates: 90 records***\n",
      "*** End dropping duplicates: 90 records***\n",
      "\n",
      "*** Start writing 90 records to staging_data/class_of_admission_etl_raw ***\n",
      "*** End writing 90 records to staging_data/class_of_admission_etl_raw ***\n",
      "\n",
      "### End class of admission extraction and transformation ###\n",
      "\n",
      "Checking for nulls\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>code</th>\n",
       "      <th>description</th>\n",
       "      <th>category</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   code  description  category\n",
       "0     0            0         0"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>code</th><th>description</th><th>category</th></tr>\n",
       "<tr><td>E3D</td><td>Spouse or Child of E3</td><td>Temporary worker visas</td></tr>\n",
       "<tr><td>U2</td><td>Spouse or U1</td><td>Other Visas</td></tr>\n",
       "<tr><td>F2</td><td>Spouses and children of F1</td><td>Student And Exchange visas</td></tr>\n",
       "<tr><td>H1C</td><td>Registered nurses participating in the Nursing ...</td><td>Temporary worker visas</td></tr>\n",
       "<tr><td>V3</td><td>Dependents of V1 or V2, visa pending</td><td>Other Visas</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+----+--------------------------------------------------+--------------------------+\n",
       "|code|                                       description|                  category|\n",
       "+----+--------------------------------------------------+--------------------------+\n",
       "| E3D|                             Spouse or Child of E3|    Temporary worker visas|\n",
       "|  U2|                                      Spouse or U1|               Other Visas|\n",
       "|  F2|                        Spouses and children of F1|Student And Exchange visas|\n",
       "| H1C|Registered nurses participating in the Nursing ...|    Temporary worker visas|\n",
       "|  V3|              Dependents of V1 or V2, visa pending|               Other Visas|\n",
       "+----+--------------------------------------------------+--------------------------+"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_class_of_admission = class_of_admission_etl(\"non_immigrant_class_of_admission.csv\", \"staging_data/class_of_admission_etl_raw\", print)\n",
    "print(\"\\nChecking for nulls\")\n",
    "display(df_class_of_admission.select(\n",
    "    [F.count(F.when(F.isnull(c), c)).alias(c) for c in df_class_of_admission.columns]\n",
    ").toPandas())\n",
    "df_class_of_admission.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- code: string (nullable = true)\n",
      " |-- description: string (nullable = true)\n",
      " |-- category: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_class_of_admission.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### US Cities Demographics\n",
    "\n",
    "###### Load and basic data filtering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "def us_city_demographics_etl(input_path=\"us-cities-demographics.csv\", output_path=\"staging_data/us_city_demographics_etl_raw\", logger=(lambda message: None)):\n",
    "    \"\"\"\n",
    "    Function that performs extraction and transformation of the US city demographics data returning a raw dataframe\n",
    "    \n",
    "    :param input_path: path containing the US city demographics data in csv format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"       \n",
    "    \n",
    "    logger(\"\\n### Starting US city demographics extraction and transformation ###\")\n",
    "    df = spark.read.csv(input_path, sep=\";\", header=True)\n",
    "            \n",
    "    logger(f\"\\n*** Start dropping duplicates: {df.count()} records***\")\n",
    "    df = df.dropDuplicates()\n",
    "    logger(f\"*** End dropping duplicates: {df.count()} records***\")\n",
    "\n",
    "    logger(f\"\\n*** Start transforming to race ratio: {df.count()} source records ***\")\n",
    "    df = df.groupBy(\n",
    "        \"State\", \"City\"\n",
    "    ).pivot(\"Race\").agg(\n",
    "        F.sum(F.col(\"Count\")).cast(\"long\").alias(\"population_count\")\n",
    "    ).select(\n",
    "        F.col(\"State\").alias(\"state\"), \n",
    "        F.col(\"City\").alias(\"city\"),\n",
    "        F.col(\"American Indian and Alaska Native\").alias(\"american_natives\"),\n",
    "        F.col(\"Asian\").alias(\"asian\"),\n",
    "        F.col(\"Black or African-American\").alias(\"african_amerian\"),\n",
    "        F.col(\"Hispanic or Latino\").alias(\"hispanic_latino\"),\n",
    "        F.col(\"White\").alias(\"white\")\n",
    "    ).orderBy(\"State\", \"City\")\n",
    "    logger(f\"*** End transforming to race ratio: {df.count()} result records  ***\")\n",
    "\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")     \n",
    "    \n",
    "    logger(\"\\n### End US city demographics extraction and transformation ###\")\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "### Starting US city demographics extraction and transformation ###\n",
      "\n",
      "*** Start dropping duplicates: 2891 records***\n",
      "*** End dropping duplicates: 2891 records***\n",
      "\n",
      "*** Start transforming to race ratio: 2891 source records ***\n",
      "*** End transforming to race ratio: 596 result records  ***\n",
      "\n",
      "*** Start writing 596 records to staging_data/us_city_demographics_etl_raw ***\n",
      "*** End writing 596 records to staging_data/us_city_demographics_etl_raw ***\n",
      "\n",
      "### End US city demographics extraction and transformation ###\n",
      "\n",
      "Checking for nulls\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>state</th>\n",
       "      <th>city</th>\n",
       "      <th>american_natives</th>\n",
       "      <th>asian</th>\n",
       "      <th>african_amerian</th>\n",
       "      <th>hispanic_latino</th>\n",
       "      <th>white</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>57</td>\n",
       "      <td>13</td>\n",
       "      <td>12</td>\n",
       "      <td>0</td>\n",
       "      <td>7</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   state  city  american_natives  asian  african_amerian  hispanic_latino  \\\n",
       "0      0     0                57     13               12                0   \n",
       "\n",
       "   white  \n",
       "0      7  "
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>state</th><th>city</th><th>american_natives</th><th>asian</th><th>african_amerian</th><th>hispanic_latino</th><th>white</th></tr>\n",
       "<tr><td>Alabama</td><td>Birmingham</td><td>1319</td><td>1500</td><td>157985</td><td>8940</td><td>51728</td></tr>\n",
       "<tr><td>Alabama</td><td>Dothan</td><td>656</td><td>1175</td><td>23243</td><td>1704</td><td>43516</td></tr>\n",
       "<tr><td>Alabama</td><td>Hoover</td><td>null</td><td>4759</td><td>18191</td><td>3430</td><td>61869</td></tr>\n",
       "<tr><td>Alabama</td><td>Huntsville</td><td>1755</td><td>6566</td><td>61561</td><td>10887</td><td>121904</td></tr>\n",
       "<tr><td>Alabama</td><td>Mobile</td><td>2816</td><td>5518</td><td>96397</td><td>5229</td><td>93755</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+-------+----------+----------------+-----+---------------+---------------+------+\n",
       "|  state|      city|american_natives|asian|african_amerian|hispanic_latino| white|\n",
       "+-------+----------+----------------+-----+---------------+---------------+------+\n",
       "|Alabama|Birmingham|            1319| 1500|         157985|           8940| 51728|\n",
       "|Alabama|    Dothan|             656| 1175|          23243|           1704| 43516|\n",
       "|Alabama|    Hoover|            null| 4759|          18191|           3430| 61869|\n",
       "|Alabama|Huntsville|            1755| 6566|          61561|          10887|121904|\n",
       "|Alabama|    Mobile|            2816| 5518|          96397|           5229| 93755|\n",
       "+-------+----------+----------------+-----+---------------+---------------+------+"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_us_city_demographics = us_city_demographics_etl(\n",
    "    \"us-cities-demographics.csv\", \"staging_data/us_city_demographics_etl_raw\", print\n",
    ")\n",
    "print(\"\\nChecking for nulls\")\n",
    "display(df_us_city_demographics.select(\n",
    "    [F.count(F.when(F.isnull(c), c)).alias(c) for c in df_us_city_demographics.columns]\n",
    ").toPandas())\n",
    "df_us_city_demographics.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- state: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- american_natives: long (nullable = true)\n",
      " |-- asian: long (nullable = true)\n",
      " |-- african_amerian: long (nullable = true)\n",
      " |-- hispanic_latino: long (nullable = true)\n",
      " |-- white: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_us_city_demographics.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Define the Data Model\n",
    "#### 3.1 Conceptual Data Model\n",
    "\n",
    "The star model is chosen in the implementation providing the following:\n",
    "\n",
    "- Query performance. The implementation uses 1 Fact data with direct joins to the dimension tables creating clear and simple join paths providing faster query performance compared to an OLTP system\n",
    "- Model is easily understood by consumer of data\n",
    "\n",
    "![Image of Conceptual Data Model](capstone-dend.png)\n",
    "\n",
    "\n",
    "#### 3.2 Mapping Out Data Pipelines\n",
    "\n",
    "- Transform and Load class of admission\n",
    "- Transform and Load airline\n",
    "- Transform and Load US city demographics\n",
    "- Transform and Load Time\n",
    "- Transform and Load Immigration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4: Run Pipelines to Model the Data \n",
    "#### 4.1 Create the data model\n",
    "Build the data pipelines to create the data model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_class_of_admission(input_path, output_path, logger):\n",
    "    \"\"\"\n",
    "    Function that performs takes in a path to the class of admission raw data transforms and loads it into the dimension version\n",
    "    \n",
    "    :param input_path: path containing the class of admission data in raw format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"           \n",
    "    df = spark.read.parquet(input_path)\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.select(\n",
    "        F.col(\"code\").alias(\"class_of_admission_id\"), \n",
    "        F.col(\"description\"),\n",
    "        F.col(\"category\")   \n",
    "    ).write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")     \n",
    "\n",
    "def create_airline(input_path, output_path, logger):\n",
    "    \"\"\"\n",
    "    Function that performs takes in a path to the airline raw data transforms and loads it into the dimension version\n",
    "    \n",
    "    :param input_path: path containing the airline data in raw format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"           \n",
    "    df = spark.read.parquet(input_path)\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.select(\n",
    "        F.col(\"iata\").alias(\"airline_id\"), \n",
    "        F.col(\"name\"),\n",
    "        F.col(\"alias\"),\n",
    "        F.col(\"callsign\"),\n",
    "        F.col(\"country\")\n",
    "    ).write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")  \n",
    "\n",
    "def create_us_city_demographic(input_path, output_path, logger):\n",
    "    \"\"\"\n",
    "    Function that performs takes in a path to the US city demographics raw data transforms and loads it into the dimension version\n",
    "    \n",
    "    :param input_path: path containing the US city demographics data in raw format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"           \n",
    "    df = spark.read.parquet(input_path)\n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.select(\n",
    "        F.col(\"state\"), \n",
    "        F.col(\"city\"),\n",
    "        F.col(\"american_natives\"),\n",
    "        F.col(\"asian\"),\n",
    "        F.col(\"african_amerian\"),\n",
    "        F.col(\"hispanic_latino\"),\n",
    "        F.col(\"white\")\n",
    "    ).write.mode('overwrite').parquet(output_path)   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\")  \n",
    "    \n",
    "def create_time(input_path, output_path, logger):\n",
    "    \"\"\"\n",
    "    Function that performs takes in a path to the immigration raw data transforms and loads it into the dimension version\n",
    "    \n",
    "    :param input_path: path containing the immigration data in raw format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"           \n",
    "    df = spark.read.parquet(input_path)\n",
    "    df_event_dates = df.select(\n",
    "        F.col(\"arrival_date_source\").alias(\"event_date_source\")\n",
    "    ).union(\n",
    "        df.select(\n",
    "            F.col(\"departure_date_source\").alias(\"event_date_source\")\n",
    "        )\n",
    "    ).distinct().where(\n",
    "        F.isnull(\"event_date_source\") == False\n",
    "    ).withColumn(\n",
    "        \"event_date\", F.to_timestamp(F.col(\"event_date_source\"), \"yyyyMMdd\")\n",
    "    ).withColumn(\n",
    "        \"day\", F.dayofmonth(\"event_date_source\")\n",
    "    ).withColumn(\n",
    "        \"week\", F.weekofyear(\"event_date_source\")\n",
    "    ).withColumn(\n",
    "        \"month\", F.month(\"event_date_source\")\n",
    "    ).withColumn(\n",
    "        \"year\", F.year(\"event_date_source\")\n",
    "    ) .withColumn(\n",
    "        \"weekday\", F.dayofweek(\"event_date_source\")\n",
    "    )\n",
    "    df_event_dates = df_event_dates.drop(\"event_date_source\")\n",
    "    logger(f\"\\n*** Start writing {df_event_dates.count()} records to {output_path} ***\")    \n",
    "    df_event_dates.write.mode('overwrite').parquet(\n",
    "        output_path, partitionBy=[ \"year\", \"month\"]\n",
    "    )   \n",
    "    logger(f\"*** End writing {df_event_dates.count()} records to {output_path} ***\")  \n",
    "\n",
    "def create_immigration(country_input_path, port_input_path, immigration_input_path, output_path, logger):\n",
    "    \"\"\"\n",
    "    Function that performs takes in a path to the immigration raw data transforms and loads it into the fact version\n",
    "    \n",
    "    :param country_input_path: path containing the I94 country code data in raw format to extract and transform\n",
    "    :param port_input_path: path containing the I94 port data in raw format to extract and transform\n",
    "    :param immigration_input_path: path containing the immigration data in raw format to extract and transform\n",
    "    :param logger: callback for logging\n",
    "    \"\"\"           \n",
    "    \n",
    "    logger(f\"\\n*** Start augmenting dataframe ***\")\n",
    "    df_country = spark.read.parquet(country_input_path)\n",
    "    df_port = spark.read.parquet(port_input_path)\n",
    "    df = spark.read.parquet(immigration_input_path)\n",
    "\n",
    "    columns = df.columns\n",
    "    columns.append(\"country as location_of_origination\")\n",
    "    df = df.join(\n",
    "        df_country, df.immigrant_city == df_country.code, how=\"left\"\n",
    "    ).selectExpr(columns)\n",
    "    \n",
    "    columns.append(\"country as location_of_residence\")\n",
    "    df = df.join(\n",
    "        df_country, df.immigrant_residence == df_country.code, how=\"left\"\n",
    "    ).selectExpr(columns)\n",
    "    \n",
    "    columns.append(\"city as arrival_city\")\n",
    "    columns.append(\"state as arrival_state\")\n",
    "    columns.append(\"country as arrival_country\")\n",
    "    df = df.join(\n",
    "        df_port, df.arrival_port == df_port.code, how='left'\n",
    "    ).selectExpr(columns)\n",
    "    columns.append(\"city as us_city\")\n",
    "    columns.append(\"state as us_state\")\n",
    "    columns.append(\"country as us_country\")\n",
    "    df = df.join(\n",
    "        df_port, df.us_residence == df_port.code, how='left'\n",
    "    ).selectExpr(columns)   \n",
    "    \n",
    "    df = df.withColumn(\n",
    "        \"arrival_date\", F.to_timestamp(F.col(\"arrival_date_source\"), \"yyyyMMdd\")\n",
    "    ).withColumn(\n",
    "        \"departure_date\", F.to_timestamp(F.col(\"departure_date_source\"), \"yyyyMMdd\")\n",
    "    )  \n",
    "    logger(f\"*** End augmenting dataframe ***\")\n",
    "    \n",
    "    logger(f\"\\n*** Start writing {df.count()} records to {output_path} ***\")\n",
    "    df.select(\n",
    "        F.col(\"admission_number\").alias(\"immigration_id\"),\n",
    "        F.col(\"location_of_origination\"),\n",
    "        F.col(\"location_of_residence\"),\n",
    "        F.col(\"arrival_date\"),\n",
    "        F.col(\"arrival_city\"),\n",
    "        F.col(\"arrival_state\"),\n",
    "        F.col(\"arrival_country\"),\n",
    "        F.col(\"us_city\"),\n",
    "        F.col(\"us_state\"),\n",
    "        F.col(\"us_country\"),\n",
    "        F.col(\"departure_date\"),\n",
    "        F.col(\"age\"),\n",
    "        F.col(\"gender\"),\n",
    "        F.col(\"airline\").alias(\"airline_id\")\n",
    "    ).write.mode('overwrite').parquet(\n",
    "        output_path, partitionBy=[ \"arrival_date\" ]\n",
    "    )   \n",
    "    logger(f\"*** End writing {df.count()} records to {output_path} ***\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "*** Start writing 90 records to model_data/class_of_admission ***\n",
      "*** End writing 90 records to model_data/class_of_admission ***\n",
      "\n",
      "*** Start writing 1120 records to model_data/airline ***\n",
      "*** End writing 1120 records to model_data/airline ***\n",
      "\n",
      "*** Start writing 596 records to model_data/us_city_demographics ***\n",
      "*** End writing 596 records to model_data/us_city_demographics ***\n",
      "\n",
      "*** Start writing 200 records to model_data/time ***\n",
      "*** End writing 200 records to model_data/time ***\n",
      "\n",
      "*** Start augmenting dataframe ***\n",
      "*** End augmenting dataframe ***\n",
      "\n",
      "*** Start writing 2994495 records to model_data/immigration ***\n",
      "*** End writing 2994495 records to model_data/immigration ***\n"
     ]
    }
   ],
   "source": [
    "create_class_of_admission(\n",
    "    \"staging_data/class_of_admission_etl_raw\", \"model_data/class_of_admission\", print\n",
    ")\n",
    "\n",
    "create_airline(\n",
    "    \"staging_data/airline_etl_raw\", \"model_data/airline\", print\n",
    ")\n",
    "\n",
    "create_us_city_demographic(\n",
    "    \"staging_data/us_city_demographics_etl_raw\", \"model_data/us_city_demographics\", print\n",
    ")\n",
    "\n",
    "create_time(\n",
    "    \"staging_data/immigration_etl_raw\", \"model_data/time\", print\n",
    ")\n",
    "\n",
    "create_immigration(\n",
    "    \"staging_data/country_etl_raw\",\n",
    "    \"staging_data/port_etl_raw\",\n",
    "    \"staging_data/immigration_etl_raw\", \n",
    "    \"model_data/immigration\", print\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "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": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "********************************************\n",
      "Start class_of_admission validation\n",
      "End class_of_admission validation successfully\n",
      "\n",
      "********************************************\n",
      "Start airline validation\n",
      "End airline validation successfully\n",
      "\n",
      "********************************************\n",
      "Start us_city_demographics validation\n",
      "End us_city_demographics validation successfully\n",
      "\n",
      "********************************************\n",
      "Start time validation\n",
      "End time validation successfully\n"
     ]
    }
   ],
   "source": [
    "# Perform quality checks here\n",
    "def validateEmptyData(df, target):\n",
    "    if df.count == 0:\n",
    "        raise ValueError(f\"{target} fails empty data check.\")\n",
    "    return f\"{target} passes empty data check.\"\n",
    "\n",
    "def validateUniqueColumn(df, target, columns):\n",
    "    result = df.groupBy(\n",
    "        columns\n",
    "    ).count().where(\n",
    "        F.col('count') > 1\n",
    "    ).select(\n",
    "        F.sum('count').alias(\"record_count\")\n",
    "    ).first()\n",
    "    if (result != None) & (result[0] != None):\n",
    "        if result[0] > 0:\n",
    "            raise ValueError(f\"{target} fails unique data check.\")\n",
    "    return f\"{target} passes unique data check.\"\n",
    "\n",
    "def validateNotNullColumn(df, target, columns):\n",
    "    result = df.select(\n",
    "        [F.count(F.when(F.isnull(column), column)).alias(column) for column in columns]\n",
    "    ).first()\n",
    "    \n",
    "    if (result != None):\n",
    "        total_count = 0\n",
    "        for count in result:\n",
    "            total_count += count\n",
    "         \n",
    "        if total_count > 0:\n",
    "            raise ValueError(f\"{target} fails not null data check.\")\n",
    "    return f\"{target} passes not null data check.\"\n",
    "\n",
    "def validateTable(df, target, columns):  \n",
    "    print(f\"\\n********************************************\")\n",
    "    print(f\"Start {target} validation\")\n",
    "    \n",
    "    assert validateEmptyData(\n",
    "            df, target\n",
    "    ) == f\"{target} passes empty data check.\"\n",
    "    \n",
    "    assert validateUniqueColumn(\n",
    "        df, target, columns  \n",
    "    ) == f\"{target} passes unique data check.\"  \n",
    "    \n",
    "    assert validateNotNullColumn(\n",
    "        df, target, columns   \n",
    "    ) == f\"{target} passes not null data check.\"      \n",
    "    \n",
    "    print(f\"End {target} validation successfully\")\n",
    "\n",
    "validateTable(\n",
    "    spark.read.parquet(\"model_data/class_of_admission\"), \n",
    "    \"class_of_admission\", \n",
    "    [\"class_of_admission_id\"]\n",
    ")\n",
    "\n",
    "validateTable(\n",
    "    spark.read.parquet(\"model_data/airline\"), \n",
    "    \"airline\", \n",
    "    [\"airline_id\"]\n",
    ")\n",
    "\n",
    "validateTable(\n",
    "    spark.read.parquet(\"model_data/us_city_demographics\"), \n",
    "    \"us_city_demographics\", \n",
    "    [\"state\", \"city\"]\n",
    ")\n",
    "\n",
    "validateTable(\n",
    "    spark.read.parquet(\"model_data/time\"), \n",
    "    \"time\", \n",
    "    [\"event_date\"]\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "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.\n",
    "\n",
    "1. airline (Dimension): Information about the airlines obtained from openflights.org. Can be used to compare travel patterns in relation to home country of airlines.\n",
    "```\n",
    " |-- airline_id: string (nullable = true), primary key representing the 2 digit IATA code\n",
    " |-- name: string (nullable = true), name of airline\n",
    " |-- alias: string (nullable = true), alias of airline used in the industry\n",
    " |-- callsign: string (nullable = true), callsign of airline used in the industry\n",
    " |-- country: string (nullable = true), home country of airline\n",
    "```\n",
    "2. class_of_admission (Dimension): Type of Visa used from entry. This data is sourced from the USCIS site. Can be used to looked into the patterns of travel for different visa holders like tourist, study, work and other needs.\n",
    "```\n",
    " |-- class_of_admission_id: string (nullable = true), code representing the Visa\n",
    " |-- description: string (nullable = true), user friendly description\n",
    " |-- category: string (nullable = true), cateogries the visa belongs to\n",
    "```\n",
    "3. time (Dimension): unique time of the events happening in the immigration Fact table and is source from the I94 Immigration Data. Can be used to look into the data broken into time periods\n",
    "```\n",
    " |-- event_date: timestamp (nullable = true), event date that is obtained from arrdate or depdate\n",
    " |-- day: integer (nullable = true), day in the month of event date\n",
    " |-- week: integer (nullable = true), week in the year of event date\n",
    " |-- weekday: integer (nullable = true), day in the week of event date\n",
    " |-- year: integer (nullable = true), year of event date\n",
    " |-- month: integer (nullable = true), month of event date\n",
    "```\n",
    "4. demographic (Dimension): US city demographics that can be used to obtained the race percentage of the city and states and find the relation of the number of vistors with their place of origin\n",
    "```\n",
    " |-- state: string (nullable = true), 2 character state code\n",
    " |-- city: string (nullable = true), name of the city\n",
    " |-- american_natives: long (nullable = true), number of Native Americans or Alaska Native\n",
    " |-- asian: long (nullable = true), number of Asians\n",
    " |-- african_amerian: long (nullable = true), number of African Americans\n",
    " |-- hispanic_latino: long (nullable = true), number of Hispanice and Latino\n",
    " |-- white: long (nullable = true), number of Whites\n",
    "```\n",
    "5. immigration (Fact): Solves the travel patterns of foreigner to United States, It also contains Gender, Age that can be used for differnt analysis\n",
    "```\n",
    " |-- immigration_id: long (nullable = true), USCIS identifier to record the entry\n",
    " |-- location_of_origination: string (nullable = true), place where foreigner comes from\n",
    " |-- location_of_residence: string (nullable = true), place where foreigner is residing in\n",
    " |-- arrival_city: string (nullable = true), city of arrival of the immigrant\n",
    " |-- arrival_state: string (nullable = true), state of arrival of the immigrant\n",
    " |-- arrival_country: string (nullable = true), country of arrival of the immigrant\n",
    " |-- us_city: string (nullable = true), city of stay of the immigrant\n",
    " |-- us_state: string (nullable = true), state of stay of the immigrant\n",
    " |-- us_country: string (nullable = true), country of stay of the immigrant\n",
    " |-- departure_date: timestamp (nullable = true), date the foriegner left the country\n",
    " |-- age: integer (nullable = true), age of the immigrant\n",
    " |-- gender: string (nullable = true), gender of the immigrant\n",
    " |-- airline_id: string (nullable = true), 2 digit IATA code of airlines\n",
    " |-- arrival_date: timestamp (nullable = true), arrival date of the immigrant\n",
    "```\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Step 5: Complete Project Write Up\n",
    "* Clearly state the rationale for the choice of tools and technologies for the project.\n",
    "\n",
    "The scripts are broken into mini functions that do not access global state. This allows them to be moved easily to the Cloud and used in other systems. The functions are also generic in taking in an input and writing out the output as blob files. This allows data to be move around easily. The same technique allows Airflow to be added in later with lesser effort since the input and output can placed in an accessible location like S3, GCS or Azure. All scripts are written using Apache Spark which allowas easily pathway into processing with Cloud based services"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Propose how often the data should be updated and why.\n",
    "\n",
    "airline and class of submission does not change frequently, so it can go on a Bi-weekly or monthly. \n",
    "The data is also pretty static and small, so it can be a full upload. \n",
    "This means the old data will be truncated with the new batch loading everything from scratch.\n",
    "\n",
    "demographic the period of update is very low so this should go on a 6 motnh to a year for the frequency.\n",
    "This will be an append update since there should little to none updates of the previous year or period's numbers.\n",
    "\n",
    "immigration and time this tables come from the freqently and should be uploaded on a shorter schedule like daily of Weekly.\n",
    "Since the data is big in size, the approached it to append to exisitng table instead of reloading everything every time\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}