Capstone Project Template.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# US Visitors DW\n",
    "__Supporting officials' decision-making to provide better visitors experience in the US__\n",
    "***\n",
    "## Overview\n",
    "The purpose of this data engineering capstone project is to give students a chance to combine what they've learned throughout the program. This project will be an important part of learners portfolio that will help to achieve data engineering-related career goals. We could choose to complete the project provided by the Udacity team or define the scope and data ourselves. I took the first approach in building the DW on the data on immigration to the United States provided by Udacity.\n",
    "\n",
    "## Business Scenario\n",
    "We are D2I (Data to Insights), a business consulting firm specialized in data warehouse services through assisting the enterprises with navigating their data needs and creating strategic operational solutions that deliver tangible business results. Specifically, we can help with the modernization of corporations' data warehousing infrastructure by improving performance and ease of use for end users, enhancing functionality, decreasing total cost of ownership while making it possible for real-time decision making. In total, our full suite of services includes helping enterprises with data profiling, data standardization, data acquisition, data transformation and integration.\n",
    "\n",
    "We have been contracted by the U.S. Customs and Border Protection to help them see what is hidden behind the data flood. We aim to model and create a brand new analytics solution on top of the state-of-the-art technolgies available to enable them to unleash insights from data then providing better customer experiences when coming to the US.\n",
    "\n",
    "## Structure of the Project\n",
    "Following the Udacity guide for this project, we structured this documentation with steps below:\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": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Scope the Project and Gather Data\n",
    "\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",
    "### The Scope \n",
    "The main deliverable of our work here will be a data warehouse in the cloud that will support answering questions through analytics tables and dashboards. Additionally, as we developed a general source-of-truth database, the Government of the US could open the solution through a web API so backend web services could query the warehouse for information relating to international visitors.\n",
    "\n",
    "### The Data\n",
    "For this work we have used the immigration, the global temperature and demographics datasets as well as the descriptions contained in the `I94_SAS_Labels_Descriptions.SAS` file.\n",
    "\n",
    "### The Architecture\n",
    "The whole solution is cloud based on top of __Amazon Web Services (AWS)__. First, all the datasets were preprocessed with __Apache Spark__ and stored in a staging area in __AWS S3__ bucket. Then, we loaded those to a __Amazon Redshift__ cluster using an __Apache Airflow__ pipeline that transfer and check the quality of the data to finally provide our customers a data mart for their convenient analysis.\n",
    "\n",
    "![Architecture](images/architecture.png)\n",
    "\n",
    "The main information and questions a user may want to extract from the data mart would be:\n",
    "\n",
    "* Visitors by nationality.\n",
    "* Visitors by origin.\n",
    "* Visitors by airline.\n",
    "* Correlations between destination in the U.S and the source country.\n",
    "* Correlations between destination in the U.S and source climates.\n",
    "* Correlations between immigration by source region, and the source region temperature.\n",
    "* Correlations between visitor demographics, and states visited.\n",
    "\n",
    "***"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Explore and Assess the Data\n",
    "\n",
    "_To familiarize ourselves with the data provided by Udacity we have done an exhaustive exploratory data analysis ([EDA](https://en.wikipedia.org/wiki/Exploratory_data_analysis)) checking what data would be useful and what preprocessing steps we should take in order to clean, organize and join the various datasets in a meaningful data model._\n",
    "\n",
    "In the following sections we briefly describe the datasets provided and give a summarized idea on the reasons we took into consideration when deciding what data to use.\n",
    "\n",
    "__Immigration Data__\n",
    "\n",
    "For decades, U.S. immigration officers issued the I-94 Form (Arrival/Departure Record) to foreign visitors (e.g., business visitors, tourists and foreign students) who lawfully entered the United States. The I-94 was a small white paper form that a foreign visitor received from cabin crews on arrival flights and from U.S. Customs and Border Protection at the time of entry into the United States. It listed the traveler's immigration category, port of entry, data of entry into the United States, status expiration date and had a unique 11-digit identifying number assigned to it. Its purpose was to record the traveler's lawful admission to the United States.\n",
    "\n",
    "This is the main dataset and there is a file for each month of the year of 2016 available in the directory `../../data/18-83510-I94-Data-2016/` in the [SAS](https://www.sas.com/en_us/home.html) binary database storage format `sas7bdat`. Combined, the 12 datasets have got more than 40 million rows (40.790.529) and 28 columns. For most of the work we used only the month of April of 2016 which has more than three million records (3.096.313)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importing the libraries needed in this project\n",
    "import os\n",
    "import pandas as pd\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "immigration_fname = '../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat'\n",
    "immigration = pd.read_sas(immigration_fname, 'sas7bdat', encoding=\"ISO-8859-1\")"
   ]
  },
  {
   "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>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>...</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>6.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>692.0</td>\n",
       "      <td>692.0</td>\n",
       "      <td>XXX</td>\n",
       "      <td>20573.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>...</td>\n",
       "      <td>U</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1979.0</td>\n",
       "      <td>10282016</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.897628e+09</td>\n",
       "      <td>NaN</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>7.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>254.0</td>\n",
       "      <td>276.0</td>\n",
       "      <td>ATL</td>\n",
       "      <td>20551.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>AL</td>\n",
       "      <td>NaN</td>\n",
       "      <td>...</td>\n",
       "      <td>Y</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1991.0</td>\n",
       "      <td>D/S</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>3.736796e+09</td>\n",
       "      <td>00296</td>\n",
       "      <td>F1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>15.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>101.0</td>\n",
       "      <td>101.0</td>\n",
       "      <td>WAS</td>\n",
       "      <td>20545.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>MI</td>\n",
       "      <td>20691.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1961.0</td>\n",
       "      <td>09302016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>OS</td>\n",
       "      <td>6.666432e+08</td>\n",
       "      <td>93</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>16.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>101.0</td>\n",
       "      <td>101.0</td>\n",
       "      <td>NYC</td>\n",
       "      <td>20545.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>MA</td>\n",
       "      <td>20567.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1988.0</td>\n",
       "      <td>09302016</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>AA</td>\n",
       "      <td>9.246846e+10</td>\n",
       "      <td>00199</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>17.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>101.0</td>\n",
       "      <td>101.0</td>\n",
       "      <td>NYC</td>\n",
       "      <td>20545.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>MA</td>\n",
       "      <td>20567.0</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>2012.0</td>\n",
       "      <td>09302016</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>AA</td>\n",
       "      <td>9.246846e+10</td>\n",
       "      <td>00199</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 28 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   cicid   i94yr  i94mon  i94cit  i94res i94port  arrdate  i94mode i94addr  \\\n",
       "0    6.0  2016.0     4.0   692.0   692.0     XXX  20573.0      NaN     NaN   \n",
       "1    7.0  2016.0     4.0   254.0   276.0     ATL  20551.0      1.0      AL   \n",
       "2   15.0  2016.0     4.0   101.0   101.0     WAS  20545.0      1.0      MI   \n",
       "3   16.0  2016.0     4.0   101.0   101.0     NYC  20545.0      1.0      MA   \n",
       "4   17.0  2016.0     4.0   101.0   101.0     NYC  20545.0      1.0      MA   \n",
       "\n",
       "   depdate   ...     entdepu  matflag  biryear   dtaddto gender insnum  \\\n",
       "0      NaN   ...           U      NaN   1979.0  10282016    NaN    NaN   \n",
       "1      NaN   ...           Y      NaN   1991.0       D/S      M    NaN   \n",
       "2  20691.0   ...         NaN        M   1961.0  09302016      M    NaN   \n",
       "3  20567.0   ...         NaN        M   1988.0  09302016    NaN    NaN   \n",
       "4  20567.0   ...         NaN        M   2012.0  09302016    NaN    NaN   \n",
       "\n",
       "  airline        admnum  fltno visatype  \n",
       "0     NaN  1.897628e+09    NaN       B2  \n",
       "1     NaN  3.736796e+09  00296       F1  \n",
       "2      OS  6.666432e+08     93       B2  \n",
       "3      AA  9.246846e+10  00199       B2  \n",
       "4      AA  9.246846e+10  00199       B2  \n",
       "\n",
       "[5 rows x 28 columns]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "immigration.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Data Dictionary__: Here, we describe the various fields of the dataset. Some descriptions were not clear enough so we had to make assumptions about the meaning."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| CICID* | ID that uniquely identify one record in the dataset |\n",
    "| I94YR | 4 digit year |\n",
    "| I94MON | Numeric month |\n",
    "| I94CIT | 3 digit code of source city for immigration (Born country) |\n",
    "| I94RES | 3 digit code of source country for immigration (Residence country) |\n",
    "| I94PORT | Port addmitted through |\n",
    "| ARRDATE | Arrival date in the USA |\n",
    "| I94MODE | Mode of transportation (1 = Air; 2 = Sea; 3 = Land; 9 = Not reported) |\n",
    "| I94ADDR | State of arrival |\n",
    "| DEPDATE | Departure date |\n",
    "| I94BIR | Age of Respondent in Years |\n",
    "| I94VISA | Visa codes collapsed into three categories: (1 = Business; 2 = Pleasure; 3 = Student) |\n",
    "| COUNT | Used for summary statistics |\n",
    "| DTADFILE | Character Date Field |\n",
    "| VISAPOST | Department of State where where Visa was issued |\n",
    "| OCCUP | Occupation that will be performed in U.S. |\n",
    "| ENTDEPA | Arrival Flag. Whether admitted or paroled into the US |\n",
    "| ENTDEPD | Departure Flag. Whether departed, lost visa, or deceased |\n",
    "| ENTDEPU | Update Flag. Update of visa, either apprehended, overstayed, or updated to PR |\n",
    "| MATFLAG | Match flag |\n",
    "| BIRYEAR | 4 digit year of birth |\n",
    "| DTADDTO | Character date field to when admitted in the US |\n",
    "| GENDER | Gender |\n",
    "| INSNUM | INS number |\n",
    "| AIRLINE | Airline used to arrive in U.S. |\n",
    "| ADMNUM | Admission number, should be unique and not nullable |\n",
    "| FLTNO | Flight number of Airline used to arrive in U.S. |\n",
    "| VISATYPE | Class of admission legally admitting the non-immigrant to temporarily stay in U.S. |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The immigration dataset is our fact so that will be at the center of the star schema model of our data warehouse."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Global Temperature Data__\n",
    "\n",
    "There are a range of organizations that collate climate trends data. The three most cited land and ocean temperature data sets are NOAA’s MLOST, NASA’s GISTEMP and the UK’s HadCrut.\n",
    "\n",
    "The Berkeley Earth, which is affiliated with Lawrence Berkeley National Laboratory, has repackaged the data from a newer compilation put it all together. The Berkeley Earth Surface Temperature Study combines 1.6 billion temperature reports from 16 pre-existing archives. It is nicely packaged and allows for slicing into interesting subsets (for example by country). They publish the source data and the code for the transformations they applied. They also use methods that allow weather observations from shorter time series to be included, meaning fewer observations need to be thrown away.\n",
    "\n",
    "In the original dataset from [Kaggle](https://www.kaggle.com/berkeleyearth/climate-change-earth-surface-temperature-data), several files are available but in this capstone project we will be using only the `GlobalLandTemperaturesByCity`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "temperature_fname = '../../data2/GlobalLandTemperaturesByCity.csv'\n",
    "world_temperature = pd.read_csv(temperature_fname)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "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>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": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "world_temperature.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Data Dictionary__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| dt | Date in format YYYY-MM-DD |\n",
    "| AverageTemperature | Average temperature of the city in a given date |\n",
    "| City | City Name |\n",
    "| Country | Country Name |\n",
    "| Latitude | Latitude |\n",
    "| Longitude | Longitude |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The dataset provides a long period of the world's temperature (from year 1743 to 2013). However, since the immigration dataset only has data of the US National Tourism Office in the year of 2016, the vast majority of the data here seems not to be suitable. We then decided to aggregate this dataset by country, averaging the temperatures and use this reduced table to join with `lookup\\I94CIT_I94RES.csv` lookup table (extracted from `I94_SAS_Labels_Descriptions.SAS`) resulting in the COUNTRY dimension of our model.\n",
    "\n",
    "> If we had temperatures of the year 2016 we could have provided an interesting analysis crossing the two tables (immigration and temperatures) in order to see how the waves of immigration to the US relate to the changes in the temperature. But this is just unfeasible due to the different dates."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "world_temperature = world_temperature.groupby([\"Country\"]).agg({\"AverageTemperature\": \"mean\", \n",
    "                                                                        \"Latitude\": \"first\", \"Longitude\": \"first\"}).reset_index()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "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>Country</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Afghanistan</td>\n",
       "      <td>13.816497</td>\n",
       "      <td>36.17N</td>\n",
       "      <td>69.61E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Albania</td>\n",
       "      <td>15.525828</td>\n",
       "      <td>40.99N</td>\n",
       "      <td>19.17E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Algeria</td>\n",
       "      <td>17.763206</td>\n",
       "      <td>36.17N</td>\n",
       "      <td>3.98E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Angola</td>\n",
       "      <td>21.759716</td>\n",
       "      <td>12.05S</td>\n",
       "      <td>13.15E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Argentina</td>\n",
       "      <td>16.999216</td>\n",
       "      <td>39.38S</td>\n",
       "      <td>62.43W</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       Country  AverageTemperature Latitude Longitude\n",
       "0  Afghanistan           13.816497   36.17N    69.61E\n",
       "1      Albania           15.525828   40.99N    19.17E\n",
       "2      Algeria           17.763206   36.17N     3.98E\n",
       "3       Angola           21.759716   12.05S    13.15E\n",
       "4    Argentina           16.999216   39.38S    62.43W"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "world_temperature.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Airports Data__\n",
    "\n",
    "The airport codes may refer to either [IATA](https://en.wikipedia.org/wiki/IATA_airport_code) airport code, a three-letter code which is used in passenger reservation, ticketing and baggage-handling systems, or the [ICAO](https://en.wikipedia.org/wiki/ICAO_airport_code) airport code which is a four letter code used by ATC systems and for airports that do not have an IATA airport code (from wikipedia).\n",
    "\n",
    "Airport codes from around the world. Downloaded from public domain source http://ourairports.com/data/ who compiled this data from multiple different sources.\n",
    "\n",
    "`airport-codes.csv` contains the list of all airport codes, the attributes are identified in datapackage description. Some of the columns contain attributes identifying airport locations, other codes (IATA, local if exist) that are relevant to identification of an airport.\n",
    "Original source url is http://ourairports.com/data/airports.csv (stored in archive/data.csv)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "airport = pd.read_csv(\"airport-codes_csv.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "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>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": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "airport.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Data Dictionary__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| ident | Unique identifier |\n",
    "| type | Type of the airport |\n",
    "| name | Airport Name |\n",
    "| elevation_ft | Altitude of the airport |\n",
    "| continent | Continent |\n",
    "| iso_country | ISO code of the country of the airport |\n",
    "| iso_region | ISO code for the region of the airport |\n",
    "| municipality | City where the airport is located |\n",
    "| gps_code | GPS code of the airport |\n",
    "| iata_code | IATA code of the airport |\n",
    "| local_code | Local code of the airport |\n",
    "| coordinates | GPS coordinates of the airport |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We are not using the airport dataset in our model. We came to a conclusion that it did not prove to be a good source of analysis once we were not able to join this to the main table immigration. We did not find a valid and consistent key in both tables in order to cross them. None of the codes (ident, gps_code, iata_code or local_code) seemed to match the columns in the immigration fact table."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__U.S. City Demographic Data__\n",
    "\n",
    "This dataset contains information about the demographics of all US cities and census-designated places with a population greater or equal to 65,000. This data comes from the US Census Bureau's 2015 American Community Survey.\n",
    "\n",
    "This product uses the Census Bureau Data API but is not endorsed or certified by the Census Bureau."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "us_cities_demographics = pd.read_csv(\"us-cities-demographics.csv\", sep=\";\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "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>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": [
    "us_cities_demographics.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Data Dictionary__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| City | Name of the city |\n",
    "| State | US state of the city |\n",
    "| Median Age | The median of the age of the population |\n",
    "| Male Population | Number of the male population |\n",
    "| Female Population | Number of the female population |\n",
    "| Total Population | Number of the total population |\n",
    "| Number of Veterans | Number of veterans living in the city |\n",
    "| Foreign-born | Number of residents of the city that were not born in the city |\n",
    "| Average Household Size | Average size of the houses in the city |\n",
    "| State Code | Code of the state of the city |\n",
    "| Race | Race class |\n",
    "| Count | Number of individual of each race |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `US Cities Demographics` is the source of the STATE dimension in our data model. We aggregated the dataset by State and pivoted the `Race` and `Count` columns in order to make each different value of Race to be a column. That way we create a complete table of statistics that summarizes the information for every US state."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Define the Data Model\n",
    "\n",
    "_In this section of the documentation we detail the process of extract, transform and load the data from the various datasets. As me mentioned before, we are using 3 of the 4 data sources provided by the Udacity team: immigration, temperatures and demographics. Also, we extract descriptions from labels descriptions file `I94_SAS_Labels_Descriptions.SAS`_\n",
    "\n",
    "#### 3.1 Conceptual Data Model\n",
    "_Map out the conceptual data model and explain why you chose that model_\n",
    "\n",
    "The immigration dataset is the origin of the center of our model. As this represent the facts of what we want to analyse - U.S visitors from the world -, this was transformed to the fact table IMMIGRATION as represented in the schema below. We gave this data most of the focus during our modeling phase. The immigration dataset is also the data source for the DATE dimension table. We extracted all the distinct values of the columns arrdate and depdate and applied various functions to store in the table a number of attributes of a particular date: day, month, year, week of year and day of week.\n",
    "\n",
    "![Star-Schema](images/star-schema.PNG)\n",
    "\n",
    "The STATE dimension table is the result of the aggregation of the demographics dataset by the State column. Median Age, Male Population, Female Population, Total Population, Number of Veterans, Foreign-born were first aggregated by `City` using `first` function, since they are repeated accross the different rows of the same city. Then, we grouped the resulting rows by `State` applying the `sum` function in the numeric columns to make a cosolidated total in each U.S State. We needed to transform the column `Race` in order to make its different values to become different columns. We achieve this by usig the pivot function of the `pyspark` package. As a result we reached to a final structure where we have got the columns (BlackOrAfricanAmerican, White, ForeignBorn, AmericanIndianAndAlaskaNative, HispanicOrLatino, Asian, NumberVeterans, FemalePopulation, MalePopulation, TotalPopulation) for each of the states of the U.S.\n",
    "\n",
    "The COUNTRY dimention completes our star schema model. To get to the structure we see in the figure above we combined the `GlobalLandTemperaturesByCity` with the code-descriptions found in the file `I94_SAS_Labels_Descriptions.SAS` for the columns `i94cit` and `i94res` showed in the image below.\n",
    "Firstly, we extracted the key-value pairs from the `I94_SAS_Labels_Descriptions.SAS` and saved those in csv files in the `lookup` directory. Following we aggregated the temperature dataset by `City` and then by `Country`. Finally, we join the two intermediary results to form the table COUNTRY. \n",
    "\n",
    "![i94cit](images/i94cit.PNG)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.2 Mapping Out Data Pipelines\n",
    "_List the steps necessary to pipeline the data into the chosen data model_\n",
    "\n",
    "\n",
    "To accomplish all the tasks related to the preprocessing of the datasets it was developed a number of functions in a package we called `helper.etl`. There you will find different helper functions to load, select, clean, transform and store the resultind datasets in a very convenient way. The open-source framework Apache Spark was the main tool in this journey. Spark provides an interface for programming entire clusters with implicit data parallelism and fault tolerance.\n",
    "\n",
    "We concentrated all the logic of preprocessing there in order to only represent here the general steps of the ETL. This notebook here is only for document purposes whereas the actual run of the ETL takes place in the Spark in cloud-native big data platform [Amazon EMR](https://aws.amazon.com/emr/?nc1=h_ls) through the execution of the main function of the `etl` package. The documentation of the functions can be found in the docstring alongside the code of the package in `helper/etl.py` file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import the ETL package\n",
    "from helper.etl import create_spark_session, etl_immigration_data, etl_countries_data, etl_states_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create Spark session\n",
    "spark = create_spark_session()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Immigration and Date datasets\n",
    "The preprocessing of the main dataset immigration starts by loading the data from the SAS file and is completed by generating and the storing of the processed dataframes to a bucket in Amazon S3. In summary, the following tasks are performed throughout the process:\n",
    "* Loading of the immigration file into Spark dataframe. We only load useful columns as we identified them in the EDA phase. In particular we discarded the follouwing fields: 'admnum', 'biryear', 'count', 'dtaddto', 'dtadfile', 'entdepa', 'entdepd', 'entdepu', 'insnum', 'matflag', 'occup', 'visapost';\n",
    "* Though some columns were actually of Integer type, the Spark framework loaded them as double or strings. To correct this we convert those fields to the proper class;\n",
    "* The dates in the immigration dataframe are stored in SAS date format, which is a value that represents the number of days between January 1, 1960, and a specified date. We convert the dates in the dataframe to a string date format in the pattern YYYY-MM-DD;\n",
    "* We drop high missing value columns \"visapost\", \"occup\", \"entdepu\" and \"insnum\";\n",
    "* Creation of `stay` column from calculating the difference in days between the departure (depdate) and arrival (arrdate) date of the visitors. That will be useful to analyse how long is the average stay of visitors and where they tend to stay longer;\n",
    "* From the date columns arrdate and depdate we create a second dataframe DATE;\n",
    "* Save the processed immigration and date dataframes to the Amazon S3 in the parquet format;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/etl_immigration.png\" alt=\"etl_immigration\" width=\"400\"/>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Perform ETL process for the Immigration dataset generating immigration and date tables and save them in the S3 bucket indicated in the output_path parameters.\n",
    "immigration = etl_immigration_data(spark, input_path='../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat',\n",
    "                                     output_path=\"s3a://data-engineer-capstone/immigration.parquet\",\n",
    "                                     date_output_path=\"s3a://data-engineer-capstone/date.parquet\",\n",
    "                                     input_format = \"com.github.saurfang.sas.spark\", \n",
    "                                     load_size=1000, partitionBy=None, \n",
    "                                     columns_to_save = '*')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Countries dataset\n",
    "The generation of the country dataset starts by loading the data global temperature dataset as well as I94CIT_I94RES lookup table and is completed by generating and the storing of the processed dataframe to a bucket in Amazon S3. In summary, the following tasks are performed throughout the process:\n",
    "* Loading of the csv file of the global temperature and I94CIT_I94RES lookup table;\n",
    "* Aggregation of the temperatures dataset by country and rename new columns;\n",
    "* Join the two datasets;\n",
    "* Save the resulting dataset to the staging area in Amazon S3;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/etl_country.png\" alt=\"etl_country\" width=\"400\"/>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Perform ETL process for the Country table. Generating the Country table and saving it in the S3 bucket indicated in the output_path parameter.\n",
    "countries = etl_countries_data(spark, output_path=e.OUTPUT + \"country.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### States dataset\n",
    "The generation of the states dataset starts by loading the data in demographics dataset as well as I94ADDR lookup table and is completed by generating and the storing of the processed dataframe to a bucket in Amazon S3. In summary, the following tasks are performed throughout the process:\n",
    "* Loading of the csv file of the demographics and I94ADDR lookup table;\n",
    "* Aggregation of the demographics dataset by state and rename new columns;\n",
    "* Join the two datasets;\n",
    "* Save the resulting dataset to the staging area in Amazon S3;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/etl_state.png\" alt=\"etl_state\" width=\"400\"/>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Perform ETL process for the State table. Generating the State table and saving it in the S3 bucket indicated in the output_path parameter.\n",
    "states = etl_states_data(spark, output_path=e.OUTPUT + \"state.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the parquet files are saved in the S3 bucket in the AWS, those are used to load the tables of the same name in the Amazon Redshift. We create the schema by running the SQL script found in `sql/create_tables.sql`. From there, our model is ready to be explored by the customers whether through open query editor in Redshift itself or using a dashboard tool such as Tableau or Power BI."
   ]
  },
  {
   "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._\n",
    "\n",
    "The whole pipeline can be divided into two stages. The first, where we used spark to load, extracted, transform and store the provided datasets into the AWS S3 staging area. The second stage we take advantage of [Apache Airflow](https://airflow.apache.org/) to build a DAG to extract data from S3 and load them into tables of the same name in Amazon Redshift. As a final step we check the data counting checking to ensure completeness.\n",
    "\n",
    "<img src=\"images/architecture.png\" alt=\"architecture\" width=\"800\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below we show the pipeline of the second stage we developed using Apache Airflow.\n",
    "\n",
    "<img src=\"images/dag.PNG\" alt=\"dag\" width=\"800\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The code to build the Airflow pipeline is located in the folder `airflow`. There you will find the code of the DAG itself (file `capstone.py` inside folder `dags`) as well as the two custom operators built for this capstone project in folder `plugins/operators`: `stage_redshift.py` and `data_quality.py`.\n",
    "\n",
    "The custom operator `StageToRedshiftOperator` was designed to load data in [parquet](https://parquet.apache.org/) format from S3 buckets in AWS and insert the content into a table in AWS Redshift. That operator is customizable to work with different buckets and with different tables by input parameters. Then it is used in our DAG to load to Redshift both fact and dimension tables.\n",
    "\n",
    "#### 4.2 Data Quality Checks\n",
    "\n",
    "First, we load the `IMMIGRATION` fact table through the step `Immigration_Fact_Table`, which is followed by the steps to load the dimension tables `STATE`, `DATE` and `COUNTRY`, respectively `State_Dimension_Table`, `Date_Dimension_Table`, `Country_Dimension_Table` steps. All the tables have a PK constraint that uniquely identify the records and in the fact table there are FK that guarantee that values in the fact are present in the dimension tables.\n",
    "\n",
    "After completing the loading process, we perform a data quality check through the step `Data_Quality_Checks` to make sure everything was OK. In this check we verify if every table was actually loaded with count check in all the tables of the model."
   ]
  },
  {
   "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",
    "\n",
    "__Table Immigration__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| CICID | Primary Key |\n",
    "| I94YR | Year |\n",
    "| I94MON | Month |\n",
    "| I94CIT | 3 digit for the country code where the visitor was born. This is a FK to the COUNTRY dimension table |\n",
    "| I94RES | 3 digit for the country code where the visitor resides in. This is a FK to the COUNTRY dimension table |\n",
    "| ARRDATE | Arrival date in the USA. This is a FK to the DATE dimension table |\n",
    "| I94MODE | Mode of transportation (1 = Air; 2 = Sea; 3 = Land; 9 = Not reported) |\n",
    "| I94ADDR | State of arrival. This is a FK to the STATE dimension table |\n",
    "| DEPDATE | Departure date from the USA. This is a FK to the DATE dimension table |\n",
    "| I94BIR | Age of Respondent in Years |\n",
    "| I94VISA | Visa codes collapsed into three categories: (1 = Business; 2 = Pleasure; 3 = Student) |\n",
    "| BIRYEAR | 4 digit year of birth |\n",
    "| GENDER | Gender |\n",
    "| AIRLINE | Airline used to arrive in U.S. |\n",
    "| FLTNO | Flight number of Airline used to arrive in U.S. |\n",
    "| VISATYPE | Class of admission legally admitting the non-immigrant to temporarily stay in U.S. |\n",
    "| STAY | Number of days in the US |\n",
    "\n",
    "\n",
    "__Table STATE__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| Code | Primary Key. This is the code of the State as in I94ADDR lookup table |\n",
    "| State | Name of the state |\n",
    "| BlackOrAfricanAmerican | Number of residents of the race Black Or African American |\n",
    "| White | Number of residents of the race White |\n",
    "| ForeignBorn | Number of residents that born outside th United States |\n",
    "| AmericanIndianAndAlaskaNative | Number of residents of the race American Indian And Alaska Native |\n",
    "| HispanicOrLatino | Number of residents of the race Hispanic Or Latino |\n",
    "| Asian | Number of residents of the race Asian |\n",
    "| NumberVeterans | Number of residents that are war veterans |\n",
    "| FemalePopulation | Number of female population |\n",
    "| MalePopulation | Number of male population |\n",
    "| TotalPopulation | Number total of the population |\n",
    "\n",
    "\n",
    "__Table COUNTRY__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| Code | Country Code. This is the PK. |\n",
    "| Country | Country Name |\n",
    "| Temperature | Average temperature of the country between 1743 and 2013 |\n",
    "| Latitude | GPS Latitude |\n",
    "| Longitude | GPS Longitude |\n",
    "\n",
    "\n",
    "__Table DATE__\n",
    "\n",
    "| Column Name | Description |\n",
    "| :--- | :--- |\n",
    "| date | Date in the format YYYY-MM-DD. This is the PK. |\n",
    "| day | Two digit day |\n",
    "| month | Two digit month |\n",
    "| year | Four digit for the year |\n",
    "| weekofyear | The week of the year |\n",
    "| dayofweek | The day of the week |"
   ]
  },
  {
   "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 whole solution implemented here is mounted on top of cloud computing technology, AWS in particular. Because the cloud computing provides a low-cost, scalable, and highly reliable infrastructure platform in the cloud this is a natural choice for every new solution like we did here. Every service we use (S3, EMR, Redshift) has reasonable cost and is ‘pay as you go’ pricing. So we can start small and scale as our solution grows. No up-front costs involved.\n",
    "\n",
    "In particular, why we use the following services:\n",
    "\n",
    "__S3:__ Provides a relatively cheap, easy-to-use with scalability, high availability, security, and performance. This seems to be perfect to a staging area like our solution here;\n",
    "\n",
    "__Spark:__ This is simply the best framework for big data processing, with built-in modules for streaming, SQL, machine learning and graph processing. Spark provides an interface for programming entire clusters with implicit data parallelism and fault tolerance. Most of our team are pythonians and Spark has a very convenient API for python programmers to use;\n",
    "\n",
    "__EMR:__ This is a cloud-native big data platform, allowing teams to process vast amounts of data quickly, and cost-effectively at scale using Spark. EMR is easy to use, secure, elastic and low-cost. Perfect to our project;\n",
    "\n",
    "__Redshift:__ A natural and logical choice since we based all the solution in the cloud in AWS. Redshift provides a massively parallel, column-oriented data warehouse that provides easy-scale functionality. The main analytical tools have native interface to load from Redshift.\n",
    "\n",
    "\n",
    "__Propose how often the data should be updated and why__\n",
    "\n",
    "Since we receive one file per month it seems reasonable to update the model monthly.\n",
    "\n",
    "__Write a description of how you would approach the problem differently under the following scenarios:__\n",
    "\n",
    " * The data was increased by 100x:\n",
    "\n",
    "Scaling the whole pipeline should not be a problem at all. Since the whole solution is on top of Amazon cloud, that are easily scalable, the only thing we would need to do is increase the number of nodes of the clusters in EMR to hadle more data. Also, Amazon Redshift is a data warehouse that can expand to exabyte-scale;\n",
    " \n",
    "* The data populates a dashboard that must be updated on a daily basis by 7am every day.\n",
    "\n",
    "The runnig interval of the Airflow DAG could be changed to daily and scheduled to run overnight to make the data available y 7am.\n",
    " \n",
    "* The database needed to be accessed by 100+ people.\n",
    " \n",
    "Again, not a big problem. With Redshift we can make use of the feature \"elastic resize\" that enables us to add or remove nodes in an Amazon Redshift cluster in minutes. This further increases the agility to get better performance and more storage for demanding workloads, and to reduce cost during periods of low demand."
   ]
  },
  {
   "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}