Capston_Project_Solution.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Project Title¶\n",
    "\n",
    "Data Engineering Capstone Project\n",
    "Project Summary\n",
    "\n",
    "--describe your project at a high level--\n",
    "\n",
    "The project follows the follow steps:\n",
    "\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": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "import re\n",
    "import pandas as pd\n",
    "import psycopg2\n",
    "from collections import defaultdict\n",
    "from datetime import datetime, timedelta\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 1.scope the project and gather the data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "1. Datasets contain million of lines and scope to provide migrant people where they area(city,state,country) going to (destination) and population at destination area.\n",
    "2. Male and Female ratio present in areas and average temperature present there.\n",
    "3. Their Mode of travel and based on which visa type.\n",
    "\n",
    "#### Resources\n",
    "\n",
    "1. Immigration data of USA\n",
    "2. Us-cities demographic  data from opensoft\n",
    "3. Global Temparature dataset from koggle\n",
    "\n",
    "\n",
    "Mainly to prepare Analytic table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "\n",
    "### Step 2: Explore and Assess the Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Getting Temperature data \n",
    "temperature_data = '../../data2/GlobalLandTemperaturesByCity.csv'\n",
    "df = pd.read_csv(temperature_data, sep=',')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "dt                               8599212\n",
      "AverageTemperature               8235082\n",
      "AverageTemperatureUncertainty    8235082\n",
      "City                             8599212\n",
      "Country                          8599212\n",
      "Latitude                         8599212\n",
      "Longitude                        8599212\n",
      "dtype: int64\n"
     ]
    }
   ],
   "source": [
    "# print number of rows in dataset\n",
    "print(df.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>dt</th>\n",
       "      <th>AverageTemperature</th>\n",
       "      <th>AverageTemperatureUncertainty</th>\n",
       "      <th>City</th>\n",
       "      <th>Country</th>\n",
       "      <th>Latitude</th>\n",
       "      <th>Longitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1743-11-01</td>\n",
       "      <td>6.068</td>\n",
       "      <td>1.737</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1743-12-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1744-01-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1744-02-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1744-03-01</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Århus</td>\n",
       "      <td>Denmark</td>\n",
       "      <td>57.05N</td>\n",
       "      <td>10.33E</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           dt  AverageTemperature  AverageTemperatureUncertainty   City  \\\n",
       "0  1743-11-01               6.068                          1.737  Århus   \n",
       "1  1743-12-01                 NaN                            NaN  Århus   \n",
       "2  1744-01-01                 NaN                            NaN  Århus   \n",
       "3  1744-02-01                 NaN                            NaN  Århus   \n",
       "4  1744-03-01                 NaN                            NaN  Århus   \n",
       "\n",
       "   Country Latitude Longitude  \n",
       "0  Denmark   57.05N    10.33E  \n",
       "1  Denmark   57.05N    10.33E  \n",
       "2  Denmark   57.05N    10.33E  \n",
       "3  Denmark   57.05N    10.33E  \n",
       "4  Denmark   57.05N    10.33E  "
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Read Immigration data which is csv format,just a sample for now then we work on big data set using spark\n",
    "path= 'immigration_data_sample.csv'\n",
    "df_immigration_pd = pd.read_csv(path, sep=',')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Unnamed: 0</th>\n",
       "      <th>cicid</th>\n",
       "      <th>i94yr</th>\n",
       "      <th>i94mon</th>\n",
       "      <th>i94cit</th>\n",
       "      <th>i94res</th>\n",
       "      <th>i94port</th>\n",
       "      <th>arrdate</th>\n",
       "      <th>i94mode</th>\n",
       "      <th>i94addr</th>\n",
       "      <th>...</th>\n",
       "      <th>entdepu</th>\n",
       "      <th>matflag</th>\n",
       "      <th>biryear</th>\n",
       "      <th>dtaddto</th>\n",
       "      <th>gender</th>\n",
       "      <th>insnum</th>\n",
       "      <th>airline</th>\n",
       "      <th>admnum</th>\n",
       "      <th>fltno</th>\n",
       "      <th>visatype</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2027561</td>\n",
       "      <td>4084316.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>209.0</td>\n",
       "      <td>209.0</td>\n",
       "      <td>HHW</td>\n",
       "      <td>20566.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>HI</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1955.0</td>\n",
       "      <td>07202016</td>\n",
       "      <td>F</td>\n",
       "      <td>NaN</td>\n",
       "      <td>JL</td>\n",
       "      <td>5.658267e+10</td>\n",
       "      <td>00782</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2171295</td>\n",
       "      <td>4422636.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>582.0</td>\n",
       "      <td>582.0</td>\n",
       "      <td>MCA</td>\n",
       "      <td>20567.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>TX</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1990.0</td>\n",
       "      <td>10222016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>*GA</td>\n",
       "      <td>9.436200e+10</td>\n",
       "      <td>XBLNG</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>589494</td>\n",
       "      <td>1195600.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>148.0</td>\n",
       "      <td>112.0</td>\n",
       "      <td>OGG</td>\n",
       "      <td>20551.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>FL</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1940.0</td>\n",
       "      <td>07052016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>LH</td>\n",
       "      <td>5.578047e+10</td>\n",
       "      <td>00464</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2631158</td>\n",
       "      <td>5291768.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>297.0</td>\n",
       "      <td>297.0</td>\n",
       "      <td>LOS</td>\n",
       "      <td>20572.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>CA</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1991.0</td>\n",
       "      <td>10272016</td>\n",
       "      <td>M</td>\n",
       "      <td>NaN</td>\n",
       "      <td>QR</td>\n",
       "      <td>9.478970e+10</td>\n",
       "      <td>00739</td>\n",
       "      <td>B2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>3032257</td>\n",
       "      <td>985523.0</td>\n",
       "      <td>2016.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>111.0</td>\n",
       "      <td>111.0</td>\n",
       "      <td>CHM</td>\n",
       "      <td>20550.0</td>\n",
       "      <td>3.0</td>\n",
       "      <td>NY</td>\n",
       "      <td>...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>M</td>\n",
       "      <td>1997.0</td>\n",
       "      <td>07042016</td>\n",
       "      <td>F</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4.232257e+10</td>\n",
       "      <td>LAND</td>\n",
       "      <td>WT</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 29 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   Unnamed: 0      cicid   i94yr  i94mon  i94cit  i94res i94port  arrdate  \\\n",
       "0     2027561  4084316.0  2016.0     4.0   209.0   209.0     HHW  20566.0   \n",
       "1     2171295  4422636.0  2016.0     4.0   582.0   582.0     MCA  20567.0   \n",
       "2      589494  1195600.0  2016.0     4.0   148.0   112.0     OGG  20551.0   \n",
       "3     2631158  5291768.0  2016.0     4.0   297.0   297.0     LOS  20572.0   \n",
       "4     3032257   985523.0  2016.0     4.0   111.0   111.0     CHM  20550.0   \n",
       "\n",
       "   i94mode i94addr    ...     entdepu  matflag  biryear   dtaddto  gender  \\\n",
       "0      1.0      HI    ...         NaN        M   1955.0  07202016       F   \n",
       "1      1.0      TX    ...         NaN        M   1990.0  10222016       M   \n",
       "2      1.0      FL    ...         NaN        M   1940.0  07052016       M   \n",
       "3      1.0      CA    ...         NaN        M   1991.0  10272016       M   \n",
       "4      3.0      NY    ...         NaN        M   1997.0  07042016       F   \n",
       "\n",
       "  insnum airline        admnum  fltno  visatype  \n",
       "0    NaN      JL  5.658267e+10  00782        WT  \n",
       "1    NaN     *GA  9.436200e+10  XBLNG        B2  \n",
       "2    NaN      LH  5.578047e+10  00464        WT  \n",
       "3    NaN      QR  9.478970e+10  00739        B2  \n",
       "4    NaN     NaN  4.232257e+10   LAND        WT  \n",
       "\n",
       "[5 rows x 29 columns]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_immigration_pd.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>City</th>\n",
       "      <th>State</th>\n",
       "      <th>Median Age</th>\n",
       "      <th>Male Population</th>\n",
       "      <th>Female Population</th>\n",
       "      <th>Total Population</th>\n",
       "      <th>Number of Veterans</th>\n",
       "      <th>Foreign-born</th>\n",
       "      <th>Average Household Size</th>\n",
       "      <th>State Code</th>\n",
       "      <th>Race</th>\n",
       "      <th>Count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Silver Spring</td>\n",
       "      <td>Maryland</td>\n",
       "      <td>33.8</td>\n",
       "      <td>40601.0</td>\n",
       "      <td>41862.0</td>\n",
       "      <td>82463</td>\n",
       "      <td>1562.0</td>\n",
       "      <td>30908.0</td>\n",
       "      <td>2.60</td>\n",
       "      <td>MD</td>\n",
       "      <td>Hispanic or Latino</td>\n",
       "      <td>25924</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Quincy</td>\n",
       "      <td>Massachusetts</td>\n",
       "      <td>41.0</td>\n",
       "      <td>44129.0</td>\n",
       "      <td>49500.0</td>\n",
       "      <td>93629</td>\n",
       "      <td>4147.0</td>\n",
       "      <td>32935.0</td>\n",
       "      <td>2.39</td>\n",
       "      <td>MA</td>\n",
       "      <td>White</td>\n",
       "      <td>58723</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Hoover</td>\n",
       "      <td>Alabama</td>\n",
       "      <td>38.5</td>\n",
       "      <td>38040.0</td>\n",
       "      <td>46799.0</td>\n",
       "      <td>84839</td>\n",
       "      <td>4819.0</td>\n",
       "      <td>8229.0</td>\n",
       "      <td>2.58</td>\n",
       "      <td>AL</td>\n",
       "      <td>Asian</td>\n",
       "      <td>4759</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Rancho Cucamonga</td>\n",
       "      <td>California</td>\n",
       "      <td>34.5</td>\n",
       "      <td>88127.0</td>\n",
       "      <td>87105.0</td>\n",
       "      <td>175232</td>\n",
       "      <td>5821.0</td>\n",
       "      <td>33878.0</td>\n",
       "      <td>3.18</td>\n",
       "      <td>CA</td>\n",
       "      <td>Black or African-American</td>\n",
       "      <td>24437</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Newark</td>\n",
       "      <td>New Jersey</td>\n",
       "      <td>34.6</td>\n",
       "      <td>138040.0</td>\n",
       "      <td>143873.0</td>\n",
       "      <td>281913</td>\n",
       "      <td>5829.0</td>\n",
       "      <td>86253.0</td>\n",
       "      <td>2.73</td>\n",
       "      <td>NJ</td>\n",
       "      <td>White</td>\n",
       "      <td>76402</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "               City          State  Median Age  Male Population  \\\n",
       "0     Silver Spring       Maryland        33.8          40601.0   \n",
       "1            Quincy  Massachusetts        41.0          44129.0   \n",
       "2            Hoover        Alabama        38.5          38040.0   \n",
       "3  Rancho Cucamonga     California        34.5          88127.0   \n",
       "4            Newark     New Jersey        34.6         138040.0   \n",
       "\n",
       "   Female Population  Total Population  Number of Veterans  Foreign-born  \\\n",
       "0            41862.0             82463              1562.0       30908.0   \n",
       "1            49500.0             93629              4147.0       32935.0   \n",
       "2            46799.0             84839              4819.0        8229.0   \n",
       "3            87105.0            175232              5821.0       33878.0   \n",
       "4           143873.0            281913              5829.0       86253.0   \n",
       "\n",
       "   Average Household Size State Code                       Race  Count  \n",
       "0                    2.60         MD         Hispanic or Latino  25924  \n",
       "1                    2.39         MA                      White  58723  \n",
       "2                    2.58         AL                      Asian   4759  \n",
       "3                    3.18         CA  Black or African-American  24437  \n",
       "4                    2.73         NJ                      White  76402  "
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Read demographic data \n",
    "path = 'us-cities-demographics.csv'\n",
    "df_demographic_pd = pd.read_csv(path,sep=\";\")\n",
    "df_demographic_pd.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Accessing Data with spark for better performance\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# creating the spark Session\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.\\\n",
    "config(\"spark.jars.packages\",\"saurfang:spark-sas7bdat:2.0.0-s_2.11\")\\\n",
    ".enableHiveSupport().getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 2.Explore and Assess the data\n",
    "#### explore\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "##### Immigration Data \n",
    "This data comes from the US National Tourism and Trade Office. We drop all rows having null or invalid values for i94port which is know for destination area."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+--------------+-----+--------+\n",
      "|cicid| i94yr|i94mon|i94cit|i94res|i94port|arrdate|i94mode|i94addr|depdate|i94bir|i94visa|count|dtadfile|visapost|occup|entdepa|entdepd|entdepu|matflag|biryear| dtaddto|gender|insnum|airline|        admnum|fltno|visatype|\n",
      "+-----+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+--------------+-----+--------+\n",
      "|  6.0|2016.0|   4.0| 692.0| 692.0|    XXX|20573.0|   null|   null|   null|  37.0|    2.0|  1.0|    null|    null| null|      T|   null|      U|   null| 1979.0|10282016|  null|  null|   null| 1.897628485E9| null|      B2|\n",
      "|  7.0|2016.0|   4.0| 254.0| 276.0|    ATL|20551.0|    1.0|     AL|   null|  25.0|    3.0|  1.0|20130811|     SEO| null|      G|   null|      Y|   null| 1991.0|     D/S|     M|  null|   null|  3.73679633E9|00296|      F1|\n",
      "| 15.0|2016.0|   4.0| 101.0| 101.0|    WAS|20545.0|    1.0|     MI|20691.0|  55.0|    2.0|  1.0|20160401|    null| null|      T|      O|   null|      M| 1961.0|09302016|     M|  null|     OS|  6.66643185E8|   93|      B2|\n",
      "| 16.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     MA|20567.0|  28.0|    2.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 1988.0|09302016|  null|  null|     AA|9.246846133E10|00199|      B2|\n",
      "| 17.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     MA|20567.0|   4.0|    2.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 2012.0|09302016|  null|  null|     AA|9.246846313E10|00199|      B2|\n",
      "| 18.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     MI|20555.0|  57.0|    1.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 1959.0|09302016|  null|  null|     AZ|9.247103803E10|00602|      B1|\n",
      "| 19.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     NJ|20558.0|  63.0|    2.0|  1.0|20160401|    null| null|      O|      K|   null|      M| 1953.0|09302016|  null|  null|     AZ|9.247139923E10|00602|      B2|\n",
      "| 20.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     NJ|20558.0|  57.0|    2.0|  1.0|20160401|    null| null|      O|      K|   null|      M| 1959.0|09302016|  null|  null|     AZ|9.247161383E10|00602|      B2|\n",
      "| 21.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     NY|20553.0|  46.0|    2.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 1970.0|09302016|  null|  null|     AZ|9.247079603E10|00602|      B2|\n",
      "| 22.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     NY|20562.0|  48.0|    1.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 1968.0|09302016|  null|  null|     AZ|9.247848973E10|00608|      B1|\n",
      "| 23.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     NY|20671.0|  52.0|    2.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 1964.0|09302016|  null|  null|     TK|9.250139443E10|00001|      B2|\n",
      "| 24.0|2016.0|   4.0| 101.0| 101.0|    TOR|20545.0|    1.0|     MO|20554.0|  33.0|    2.0|  1.0|20160401|    null| null|      O|      O|   null|      M| 1983.0|09302016|  null|  null|     MQ|9.249090503E10|03348|      B2|\n",
      "| 27.0|2016.0|   4.0| 101.0| 101.0|    BOS|20545.0|    1.0|     MA|20549.0|  58.0|    1.0|  1.0|20160401|     TIA| null|      G|      O|   null|      M| 1958.0|04062016|     M|  null|     LH|9.247876383E10|00422|      B1|\n",
      "| 28.0|2016.0|   4.0| 101.0| 101.0|    ATL|20545.0|    1.0|     MA|20549.0|  56.0|    1.0|  1.0|20160401|     TIA| null|      G|      O|   null|      M| 1960.0|04062016|     F|  null|     LH|9.247890033E10|00422|      B1|\n",
      "| 29.0|2016.0|   4.0| 101.0| 101.0|    ATL|20545.0|    1.0|     MA|20561.0|  62.0|    2.0|  1.0|20160401|     TIA| null|      G|      O|   null|      M| 1954.0|09302016|     M|  null|     AZ|9.250378143E10|00614|      B2|\n",
      "| 30.0|2016.0|   4.0| 101.0| 101.0|    ATL|20545.0|    1.0|     NJ|20578.0|  49.0|    2.0|  1.0|20160401|     TIA| null|      G|      O|   null|      M| 1967.0|09302016|     M|  null|     OS|9.247020943E10|00089|      B2|\n",
      "| 31.0|2016.0|   4.0| 101.0| 101.0|    ATL|20545.0|    1.0|     NY|20611.0|  43.0|    2.0|  1.0|20160401|     TIA| null|      G|      O|   null|      M| 1973.0|09302016|     M|  null|     OS|9.247128923E10|00089|      B2|\n",
      "| 33.0|2016.0|   4.0| 101.0| 101.0|    HOU|20545.0|    1.0|     TX|20554.0|  53.0|    2.0|  1.0|20160401|     TIA| null|      G|      O|   null|      M| 1963.0|09302016|     F|  null|     TK|9.250930163E10|00033|      B2|\n",
      "| 34.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     CT|   null|  48.0|    2.0|  1.0|20160401|     TIA| null|      G|   null|   null|   null| 1968.0|09302016|     M|  null|     AZ|9.247042023E10|00602|      B2|\n",
      "| 35.0|2016.0|   4.0| 101.0| 101.0|    NYC|20545.0|    1.0|     CT|   null|  74.0|    2.0|  1.0|20160401|     TIA| null|      T|   null|   null|   null| 1942.0|09302016|     F|  null|     TK|  6.69712185E8|    1|      B2|\n",
      "+-----+------+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+--------+------+------+-------+--------------+-----+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "path_immigration_data= '../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat'\n",
    "spark_immigration_data = spark.read.format('com.github.saurfang.sas.spark').load(path_immigration_data)\n",
    "spark_immigration_data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Creating dictionary of i94port which are of valid format\n",
    "re_compile_obj = re.compile(r'\\'(.*)\\'.*\\'(.*)\\'')\n",
    "valid_i94port = {}\n",
    "with open('i94port_valid.txt') as f:\n",
    "     for line in f:\n",
    "            match = re_compile_obj.search(line)\n",
    "            valid_i94port[match[1]]=[match[2]]\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# clean i94 immigration data using valid ports data obtained above\n",
    "spark_immigration_valid_port = spark_immigration_data.filter(spark_immigration_data.i94port.isin(list(valid_i94port)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Temperature Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|        dt| AverageTemperature|AverageTemperatureUncertainty| City|Country|Latitude|Longitude|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|1743-11-01|              6.068|           1.7369999999999999|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1743-12-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-01-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-02-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-03-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-04-01| 5.7879999999999985|           3.6239999999999997|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-05-01|             10.644|           1.2830000000000001|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-06-01| 14.050999999999998|                        1.347|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-07-01|             16.082|                        1.396|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-08-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-09-01| 12.780999999999999|                        1.454|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-10-01|               7.95|                         1.63|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-11-01|  4.638999999999999|           1.3019999999999998|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-12-01|0.12199999999999987|                        1.756|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-01-01|-1.3330000000000002|                        1.642|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-02-01|             -2.732|                        1.358|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-03-01|              0.129|                        1.088|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-04-01|              4.042|                        1.138|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-05-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-06-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "path = \"../../data2/GlobalLandTemperaturesByCity.csv\"\n",
    "spark_temp = spark.read.format(\"csv\").option(\"header\",True).option(\"delimiter\",\",\").load(path)\n",
    "spark_temp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "#Adding a column \"i94port\" to spark_temp dataset\n",
    "\n",
    "@udf\n",
    "def get_column(city):\n",
    "    for key in valid_i94port:\n",
    "        if key.lower() in valid_i94port[key][0].lower():\n",
    "            return key\n",
    "        \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|        dt| AverageTemperature|AverageTemperatureUncertainty| City|Country|Latitude|Longitude|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|1743-11-01|              6.068|           1.7369999999999999|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1743-12-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-01-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-02-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-03-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-04-01| 5.7879999999999985|           3.6239999999999997|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-05-01|             10.644|           1.2830000000000001|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-06-01| 14.050999999999998|                        1.347|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-07-01|             16.082|                        1.396|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-08-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-09-01| 12.780999999999999|                        1.454|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-10-01|               7.95|                         1.63|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-11-01|  4.638999999999999|           1.3019999999999998|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-12-01|0.12199999999999987|                        1.756|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-01-01|-1.3330000000000002|                        1.642|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-02-01|             -2.732|                        1.358|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-03-01|              0.129|                        1.088|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-04-01|              4.042|                        1.138|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-05-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-06-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark_temp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3490"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark_temp.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8235082"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Number of records after removing null values in AverageTemperature column\n",
    "spark_temp = spark_temp.filter(spark_temp.AverageTemperature != 'null')\n",
    "spark_temp.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|        dt| AverageTemperature|AverageTemperatureUncertainty| City|Country|Latitude|Longitude|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|1743-11-01|              6.068|           1.7369999999999999|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-04-01| 5.7879999999999985|           3.6239999999999997|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-05-01|             10.644|           1.2830000000000001|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-06-01| 14.050999999999998|                        1.347|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-07-01|             16.082|                        1.396|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-09-01| 12.780999999999999|                        1.454|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-10-01|               7.95|                         1.63|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-11-01|  4.638999999999999|           1.3019999999999998|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-12-01|0.12199999999999987|                        1.756|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-01-01|-1.3330000000000002|                        1.642|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-02-01|             -2.732|                        1.358|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-03-01|              0.129|                        1.088|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-04-01|              4.042|                        1.138|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-01-01|              1.699|                        1.013|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-02-01| 3.9610000000000003|           2.3609999999999998|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-03-01|              5.182|                         3.48|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-04-01|              7.197|                        0.732|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-05-01|             10.634|                        1.351|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-06-01|             14.913|                        1.181|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1750-07-01|             17.831|                         1.22|Århus|Denmark|  57.05N|   10.33E|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark_temp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "|        dt|  AverageTemperature|AverageTemperatureUncertainty|        City|           Country|Latitude|Longitude|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "|1743-11-01|               3.264|                        1.665|   Allentown|     United States|  40.99N|   74.56W|\n",
      "|1779-11-01|0.011999999999999985|                        2.714|      Atyrau|        Kazakhstan|  47.42N|   50.92E|\n",
      "|1825-01-01|  26.069000000000003|                         2.16|     Bintulu|          Malaysia|   2.41N|  113.30E|\n",
      "|1825-01-01|              26.517|           2.5839999999999996| Butterworth|          Malaysia|   5.63N|  100.09E|\n",
      "|1845-01-01|              24.995|                        1.871|      Cainta|       Philippines|  15.27N|  120.83E|\n",
      "|1825-01-01|              24.753|           2.1519999999999997|      Ciamis|         Indonesia|   7.23S|  107.84E|\n",
      "|1850-01-01|              22.121|           1.5730000000000002|      Dodoma|          Tanzania|   5.63S|   35.52E|\n",
      "|1840-01-01|  6.2250000000000005|                        2.112|      Fuling|             China|  29.74N|  107.08E|\n",
      "|1841-01-01| -0.3360000000000001|                        2.695|      Fuyang|             China|  32.95N|  115.85E|\n",
      "|1856-01-01|              26.901|                        1.359|         Ife|           Nigeria|   7.23N|    4.05E|\n",
      "|1796-01-01|  14.427999999999999|                        2.395|  Jhunjhunun|             India|  28.13N|   75.45E|\n",
      "|1857-01-01|               25.32|           1.5130000000000001|      Maxixe|        Mozambique|  23.31S|   35.83E|\n",
      "|1856-01-01|  26.555000000000003|                        1.308|      Owerri|           Nigeria|   5.63N|    6.46E|\n",
      "|1824-01-01|  24.958000000000002|                        1.182|Puerto Plata|Dominican Republic|  20.09N|   69.95W|\n",
      "|1832-01-01|              22.288|                        1.587| Santo André|            Brazil|  23.31S|   46.31W|\n",
      "|1857-01-01|  18.581000000000003|           1.8119999999999998|      Soweto|      South Africa|  26.52S|   28.66E|\n",
      "|1825-01-01|              25.862|                        2.175|    Sukabumi|         Indonesia|   7.23S|  106.22E|\n",
      "|1743-11-01|               7.084|           1.8969999999999998|    Toulouse|            France|  44.20N|    2.24E|\n",
      "|1835-01-01|              13.086|                        1.942|  Tulancingo|            Mexico|  20.09N|   98.96W|\n",
      "|1849-01-01|  26.531999999999996|                        1.416|      Anyama|     Côte D'Ivoire|   5.63N|    4.84W|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark_temp = spark_temp.dropDuplicates(['City','Country'])\n",
    "spark_temp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3490"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark_temp.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Us-Cities-demographics\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+----------+---------------+-----------------+----------------+------------------+------------+----------------------+----------+--------------------+------+\n",
      "|            City|         State|Median Age|Male Population|Female Population|Total Population|Number of Veterans|Foreign-born|Average Household Size|State Code|                Race| Count|\n",
      "+----------------+--------------+----------+---------------+-----------------+----------------+------------------+------------+----------------------+----------+--------------------+------+\n",
      "|   Silver Spring|      Maryland|      33.8|          40601|            41862|           82463|              1562|       30908|                   2.6|        MD|  Hispanic or Latino| 25924|\n",
      "|          Quincy| Massachusetts|      41.0|          44129|            49500|           93629|              4147|       32935|                  2.39|        MA|               White| 58723|\n",
      "|          Hoover|       Alabama|      38.5|          38040|            46799|           84839|              4819|        8229|                  2.58|        AL|               Asian|  4759|\n",
      "|Rancho Cucamonga|    California|      34.5|          88127|            87105|          175232|              5821|       33878|                  3.18|        CA|Black or African-...| 24437|\n",
      "|          Newark|    New Jersey|      34.6|         138040|           143873|          281913|              5829|       86253|                  2.73|        NJ|               White| 76402|\n",
      "|          Peoria|      Illinois|      33.1|          56229|            62432|          118661|              6634|        7517|                   2.4|        IL|American Indian a...|  1343|\n",
      "|        Avondale|       Arizona|      29.1|          38712|            41971|           80683|              4815|        8355|                  3.18|        AZ|Black or African-...| 11592|\n",
      "|     West Covina|    California|      39.8|          51629|            56860|          108489|              3800|       37038|                  3.56|        CA|               Asian| 32716|\n",
      "|        O'Fallon|      Missouri|      36.0|          41762|            43270|           85032|              5783|        3269|                  2.77|        MO|  Hispanic or Latino|  2583|\n",
      "|      High Point|North Carolina|      35.5|          51751|            58077|          109828|              5204|       16315|                  2.65|        NC|               Asian| 11060|\n",
      "|          Folsom|    California|      40.9|          41051|            35317|           76368|              4187|       13234|                  2.62|        CA|  Hispanic or Latino|  5822|\n",
      "|          Folsom|    California|      40.9|          41051|            35317|           76368|              4187|       13234|                  2.62|        CA|American Indian a...|   998|\n",
      "|    Philadelphia|  Pennsylvania|      34.1|         741270|           826172|         1567442|             61995|      205339|                  2.61|        PA|               Asian|122721|\n",
      "|         Wichita|        Kansas|      34.6|         192354|           197601|          389955|             23978|       40270|                  2.56|        KS|  Hispanic or Latino| 65162|\n",
      "|         Wichita|        Kansas|      34.6|         192354|           197601|          389955|             23978|       40270|                  2.56|        KS|American Indian a...|  8791|\n",
      "|      Fort Myers|       Florida|      37.3|          36850|            37165|           74015|              4312|       15365|                  2.45|        FL|               White| 50169|\n",
      "|      Pittsburgh|  Pennsylvania|      32.9|         149690|           154695|          304385|             17728|       28187|                  2.13|        PA|               White|208863|\n",
      "|          Laredo|         Texas|      28.8|         124305|           131484|          255789|              4921|       68427|                  3.66|        TX|American Indian a...|  1253|\n",
      "|        Berkeley|    California|      32.5|          60142|            60829|          120971|              3736|       25000|                  2.35|        CA|               Asian| 27089|\n",
      "|     Santa Clara|    California|      35.2|          63278|            62938|          126216|              4426|       52281|                  2.75|        CA|               White| 55847|\n",
      "+----------------+--------------+----------+---------------+-----------------+----------------+------------------+------------+----------------------+----------+--------------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "path = 'us-cities-demographics.csv'\n",
    "spark_us_cities = spark.read.format(\"csv\").option(\"header\", True).option(\"delimiter\",\";\").load(path)\n",
    "spark_us_cities.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "596"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark_us_cities = spark_us_cities.dropDuplicates([\"City\",\"State\"])\n",
    "spark_us_cities.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 3.Define the Data Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "3.1 Conceptual Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "First dimentional model contains data from i94 immigration dataset\n",
    "- i94yr = 4 digit year\n",
    "- i94mon = numeric month\n",
    "- i94cit = 3 digit code of origin city\n",
    "- i94port = 3 character code of destination city\n",
    "- arrdate = arrival date\n",
    "- i94mode = 1 digit travel code\n",
    "- depdate = departure date\n",
    "- i94visa = reason for immigration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Second dimentional model contain data from GobalTemperature dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- City = city name\n",
    "- Country = Country name\n",
    "- latitude = latitude \n",
    "- longitude = longitude\n",
    "- avaerage temperature = Average temperature pesent in specific latitude and longitude\n",
    "- i94port = 3 character code of destination city (added using immigration data)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Third dimentional model contains data from us-cities-demographics dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- City = city name\n",
    "- State = state name\n",
    "- Male population - male population in that city\n",
    "- Female population - Female Population in that city\n",
    "- race = people of specific race\n",
    "- Average household size = avaerage house hold size"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "****Fist Fact Table1**** \n",
    "\n",
    "##### Contain information by joining immigration data to temperature data on i94port and again joing to us-cities-demog   raphics data on city name####"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- i94yr = 4 digit year code\n",
    "- i94mon =  digit month code(1,,,,12)\n",
    "- i94cit = 3 digit of origin city\n",
    "- i94port = 3 character code of destination city\n",
    "- arrdata = arrival data\n",
    "- i94mode = mode of travel\n",
    "- gender = gender\n",
    "- airline = airline code\n",
    "- deptdate = departure date\n",
    "- i94visa = type of reason for visa\n",
    "- Average Temperature= average temperature in city\n",
    "- Male Population = male population in that city\n",
    "- Female population = femle population\n",
    "- race = people of specific race\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "****Fist Fact Table2**** "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "##### contain information by joining selected columns from fact table1 and us-cities-demographics dataset on city column.Where city were only from USA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- city = city name\n",
    "- male population = male population\n",
    "- female population = female population\n",
    "- average household size = average household size\n",
    "- reason = reason\n",
    "- i94port = 3 character destional city\n",
    "- i94month = numeric month\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "****Necessary steps to pipline the data (datamodel specified above)****"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 1.Clean the immigration dataset by removing duplicates and null values in i94port port and making spark dataframe done this in above stemps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 2.Clean temperature dataset on addng i94port column with valid state code and removing duplicate state and country names."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 3.Create first dimentional table after cleaning and removing duplicates and null values in i94immigration data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 4.Create second dimentional table on global temperature dataset with column \"i94port\"."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 5.Create third dimentional table on us-cities-demographics dataset with column names specified above and removing duplicate cities and statues"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 6.Create fact table by joining immigration and temperature dimension tables on i94port and again join demographics dimentional table on city and write to parquet file partitioned by i94port and city."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "**Run ETL to Model the Data**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "***Step1***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Path to I94 immigration data \n",
    "# Here only april data is loaded but we can load whole millions of data using \n",
    "# path = '../../data/18-83510-I94-Data-2016/*.sas7bdat'\n",
    "path = '../../data/18-83510-I94-Data-2016/i94_jun16_sub.sas7bdat'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# we have already loaded april data and done some calculation , now using june data \n",
    "spark_immigration_data = spark.read.format('com.github.saurfang.sas.spark').load(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "***Cleaning Data***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Creating dictionary of i94port which are of valid format for june month\n",
    "re_compile_obj = re.compile(r'\\'(.*)\\'.*\\'(.*)\\'')\n",
    "valid_i94port = {}\n",
    "with open('i94port_valid.txt') as f:\n",
    "     for line in f:\n",
    "            match = re_compile_obj.search(line)\n",
    "            valid_i94port[match[1]]=[match[2]]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark_immigration_data = spark_immigration_data.filter(spark_immigration_data.i94port.isin(list(valid_i94port)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+\n",
      "|i94port|\n",
      "+-------+\n",
      "|    SFR|\n",
      "|    SFR|\n",
      "|    HOU|\n",
      "|    HOU|\n",
      "|    BOS|\n",
      "|    NEW|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    NYC|\n",
      "|    TAM|\n",
      "|    BOS|\n",
      "|    BOS|\n",
      "+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# selecting i94port column which contain only valid port\n",
    "spark_immigration_data.select([\"i94port\"]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------+-------+-------+-------+-------+-------+\n",
      "| i94yr|i94mon|i94cit|i94port|arrdate|i94mode|depdate|i94visa|\n",
      "+------+------+------+-------+-------+-------+-------+-------+\n",
      "|2016.0|   6.0| 254.0|    SFR|20623.0|    1.0|   null|    3.0|\n",
      "|2016.0|   6.0| 254.0|    SFR|20623.0|    1.0|   null|    3.0|\n",
      "|2016.0|   6.0| 127.0|    HOU|20617.0|    1.0|20679.0|    2.0|\n",
      "|2016.0|   6.0| 127.0|    HOU|20617.0|    1.0|20679.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    BOS|20606.0|    1.0|20608.0|    1.0|\n",
      "|2016.0|   6.0| 101.0|    NEW|20606.0|    1.0|20639.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20610.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20618.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20610.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20610.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20610.0|    1.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20610.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20626.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20615.0|    1.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20615.0|    1.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20615.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    NYC|20606.0|    1.0|20616.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    TAM|20606.0|    1.0|20662.0|    2.0|\n",
      "|2016.0|   6.0| 101.0|    BOS|20606.0|    1.0|   null|    3.0|\n",
      "|2016.0|   6.0| 101.0|    BOS|20606.0|    1.0|20632.0|    2.0|\n",
      "+------+------+------+-------+-------+-------+-------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create immigration dimentional table\n",
    "immigration_dim = spark_immigration_data.select([\"i94yr\", \"i94mon\", \"i94cit\", \"i94port\", \"arrdate\", \"i94mode\", \"depdate\", \"i94visa\"])\n",
    "immigration_dim.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# writing data to separet folder in parquet formate and partition by i94port and i94month\n",
    "immigration_dim.write.mode(\"append\").partitionBy(\"i94port\").parquet(\"/results/immigration.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "***step2***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|        dt| AverageTemperature|AverageTemperatureUncertainty| City|Country|Latitude|Longitude|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "|1743-11-01|              6.068|           1.7369999999999999|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1743-12-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-01-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-02-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-03-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-04-01| 5.7879999999999985|           3.6239999999999997|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-05-01|             10.644|           1.2830000000000001|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-06-01| 14.050999999999998|                        1.347|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-07-01|             16.082|                        1.396|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-08-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-09-01| 12.780999999999999|                        1.454|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-10-01|               7.95|                         1.63|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-11-01|  4.638999999999999|           1.3019999999999998|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1744-12-01|0.12199999999999987|                        1.756|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-01-01|-1.3330000000000002|                        1.642|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-02-01|             -2.732|                        1.358|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-03-01|              0.129|                        1.088|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-04-01|              4.042|                        1.138|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-05-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "|1745-06-01|               null|                         null|Århus|Denmark|  57.05N|   10.33E|\n",
      "+----------+-------------------+-----------------------------+-----+-------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# path to temperature data\n",
    "fname = '../../data2/GlobalLandTemperaturesByCity.csv'\n",
    "temperature_dim = spark.read.format(\"csv\").option(\"header\",True).option(\"delimiter\", \",\").load(fname)\n",
    "temperature_dim.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "|        dt|  AverageTemperature|AverageTemperatureUncertainty|        City|           Country|Latitude|Longitude|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "|1743-11-01|               3.264|                        1.665|   Allentown|     United States|  40.99N|   74.56W|\n",
      "|1779-11-01|0.011999999999999985|                        2.714|      Atyrau|        Kazakhstan|  47.42N|   50.92E|\n",
      "|1825-01-01|  26.069000000000003|                         2.16|     Bintulu|          Malaysia|   2.41N|  113.30E|\n",
      "|1825-01-01|              26.517|           2.5839999999999996| Butterworth|          Malaysia|   5.63N|  100.09E|\n",
      "|1845-01-01|              24.995|                        1.871|      Cainta|       Philippines|  15.27N|  120.83E|\n",
      "|1825-01-01|              24.753|           2.1519999999999997|      Ciamis|         Indonesia|   7.23S|  107.84E|\n",
      "|1850-01-01|              22.121|           1.5730000000000002|      Dodoma|          Tanzania|   5.63S|   35.52E|\n",
      "|1840-01-01|  6.2250000000000005|                        2.112|      Fuling|             China|  29.74N|  107.08E|\n",
      "|1841-01-01| -0.3360000000000001|                        2.695|      Fuyang|             China|  32.95N|  115.85E|\n",
      "|1856-01-01|              26.901|                        1.359|         Ife|           Nigeria|   7.23N|    4.05E|\n",
      "|1796-01-01|  14.427999999999999|                        2.395|  Jhunjhunun|             India|  28.13N|   75.45E|\n",
      "|1857-01-01|               25.32|           1.5130000000000001|      Maxixe|        Mozambique|  23.31S|   35.83E|\n",
      "|1856-01-01|  26.555000000000003|                        1.308|      Owerri|           Nigeria|   5.63N|    6.46E|\n",
      "|1824-01-01|  24.958000000000002|                        1.182|Puerto Plata|Dominican Republic|  20.09N|   69.95W|\n",
      "|1832-01-01|              22.288|                        1.587| Santo André|            Brazil|  23.31S|   46.31W|\n",
      "|1857-01-01|  18.581000000000003|           1.8119999999999998|      Soweto|      South Africa|  26.52S|   28.66E|\n",
      "|1825-01-01|              25.862|                        2.175|    Sukabumi|         Indonesia|   7.23S|  106.22E|\n",
      "|1743-11-01|               7.084|           1.8969999999999998|    Toulouse|            France|  44.20N|    2.24E|\n",
      "|1835-01-01|              13.086|                        1.942|  Tulancingo|            Mexico|  20.09N|   98.96W|\n",
      "|1849-01-01|  26.531999999999996|                        1.416|      Anyama|     Côte D'Ivoire|   5.63N|    4.84W|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "temperature_dim = temperature_dim.dropDuplicates([\"City\",\"Country\"])\n",
    "temperature_dim.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3490"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "temperature_dim.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "***Cleaning Step***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "|        dt|  AverageTemperature|AverageTemperatureUncertainty|        City|           Country|Latitude|Longitude|i94port|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "|1743-11-01|               3.264|                        1.665|   Allentown|     United States|  40.99N|   74.56W|    ALC|\n",
      "|1779-11-01|0.011999999999999985|                        2.714|      Atyrau|        Kazakhstan|  47.42N|   50.92E|    ALC|\n",
      "|1825-01-01|  26.069000000000003|                         2.16|     Bintulu|          Malaysia|   2.41N|  113.30E|    ALC|\n",
      "|1825-01-01|              26.517|           2.5839999999999996| Butterworth|          Malaysia|   5.63N|  100.09E|    ALC|\n",
      "|1845-01-01|              24.995|                        1.871|      Cainta|       Philippines|  15.27N|  120.83E|    ALC|\n",
      "|1825-01-01|              24.753|           2.1519999999999997|      Ciamis|         Indonesia|   7.23S|  107.84E|    ALC|\n",
      "|1850-01-01|              22.121|           1.5730000000000002|      Dodoma|          Tanzania|   5.63S|   35.52E|    ALC|\n",
      "|1840-01-01|  6.2250000000000005|                        2.112|      Fuling|             China|  29.74N|  107.08E|    ALC|\n",
      "|1841-01-01| -0.3360000000000001|                        2.695|      Fuyang|             China|  32.95N|  115.85E|    ALC|\n",
      "|1856-01-01|              26.901|                        1.359|         Ife|           Nigeria|   7.23N|    4.05E|    ALC|\n",
      "|1796-01-01|  14.427999999999999|                        2.395|  Jhunjhunun|             India|  28.13N|   75.45E|    ALC|\n",
      "|1857-01-01|               25.32|           1.5130000000000001|      Maxixe|        Mozambique|  23.31S|   35.83E|    ALC|\n",
      "|1856-01-01|  26.555000000000003|                        1.308|      Owerri|           Nigeria|   5.63N|    6.46E|    ALC|\n",
      "|1824-01-01|  24.958000000000002|                        1.182|Puerto Plata|Dominican Republic|  20.09N|   69.95W|    ALC|\n",
      "|1832-01-01|              22.288|                        1.587| Santo André|            Brazil|  23.31S|   46.31W|    ALC|\n",
      "|1857-01-01|  18.581000000000003|           1.8119999999999998|      Soweto|      South Africa|  26.52S|   28.66E|    ALC|\n",
      "|1825-01-01|              25.862|                        2.175|    Sukabumi|         Indonesia|   7.23S|  106.22E|    ALC|\n",
      "|1743-11-01|               7.084|           1.8969999999999998|    Toulouse|            France|  44.20N|    2.24E|    ALC|\n",
      "|1835-01-01|              13.086|                        1.942|  Tulancingo|            Mexico|  20.09N|   98.96W|    ALC|\n",
      "|1849-01-01|  26.531999999999996|                        1.416|      Anyama|     Côte D'Ivoire|   5.63N|    4.84W|    ALC|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Adding column i94port to temperature_dim table\n",
    "temperature_dim = temperature_dim.withColumn(\"i94port\", get_column(temperature_dim.City))\n",
    "temperature_dim.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "|        dt|  AverageTemperature|AverageTemperatureUncertainty|        City|           Country|Latitude|Longitude|i94port|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "|1743-11-01|               3.264|                        1.665|   Allentown|     United States|  40.99N|   74.56W|    ALC|\n",
      "|1779-11-01|0.011999999999999985|                        2.714|      Atyrau|        Kazakhstan|  47.42N|   50.92E|    ALC|\n",
      "|1825-01-01|  26.069000000000003|                         2.16|     Bintulu|          Malaysia|   2.41N|  113.30E|    ALC|\n",
      "|1825-01-01|              26.517|           2.5839999999999996| Butterworth|          Malaysia|   5.63N|  100.09E|    ALC|\n",
      "|1845-01-01|              24.995|                        1.871|      Cainta|       Philippines|  15.27N|  120.83E|    ALC|\n",
      "|1825-01-01|              24.753|           2.1519999999999997|      Ciamis|         Indonesia|   7.23S|  107.84E|    ALC|\n",
      "|1850-01-01|              22.121|           1.5730000000000002|      Dodoma|          Tanzania|   5.63S|   35.52E|    ALC|\n",
      "|1840-01-01|  6.2250000000000005|                        2.112|      Fuling|             China|  29.74N|  107.08E|    ALC|\n",
      "|1841-01-01| -0.3360000000000001|                        2.695|      Fuyang|             China|  32.95N|  115.85E|    ALC|\n",
      "|1856-01-01|              26.901|                        1.359|         Ife|           Nigeria|   7.23N|    4.05E|    ALC|\n",
      "|1796-01-01|  14.427999999999999|                        2.395|  Jhunjhunun|             India|  28.13N|   75.45E|    ALC|\n",
      "|1857-01-01|               25.32|           1.5130000000000001|      Maxixe|        Mozambique|  23.31S|   35.83E|    ALC|\n",
      "|1856-01-01|  26.555000000000003|                        1.308|      Owerri|           Nigeria|   5.63N|    6.46E|    ALC|\n",
      "|1824-01-01|  24.958000000000002|                        1.182|Puerto Plata|Dominican Republic|  20.09N|   69.95W|    ALC|\n",
      "|1832-01-01|              22.288|                        1.587| Santo André|            Brazil|  23.31S|   46.31W|    ALC|\n",
      "|1857-01-01|  18.581000000000003|           1.8119999999999998|      Soweto|      South Africa|  26.52S|   28.66E|    ALC|\n",
      "|1825-01-01|              25.862|                        2.175|    Sukabumi|         Indonesia|   7.23S|  106.22E|    ALC|\n",
      "|1743-11-01|               7.084|           1.8969999999999998|    Toulouse|            France|  44.20N|    2.24E|    ALC|\n",
      "|1835-01-01|              13.086|                        1.942|  Tulancingo|            Mexico|  20.09N|   98.96W|    ALC|\n",
      "|1849-01-01|  26.531999999999996|                        1.416|      Anyama|     Côte D'Ivoire|   5.63N|    4.84W|    ALC|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Removing all null values in Average Temperature column\n",
    "temperature_dim = temperature_dim.filter(temperature_dim[\"AverageTemperature\"] != 'null')\n",
    "temperature_dim.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Removing all duplicates in city and country columns \n",
    "temperature_dim = temperature_dim.dropDuplicates([\"City\",\"Country\"])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "|        dt|  AverageTemperature|AverageTemperatureUncertainty|        city|           Country|Latitude|Longitude|i94port|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "|1743-11-01|               3.264|                        1.665|   Allentown|     United States|  40.99N|   74.56W|    ALC|\n",
      "|1779-11-01|0.011999999999999985|                        2.714|      Atyrau|        Kazakhstan|  47.42N|   50.92E|    ALC|\n",
      "|1825-01-01|  26.069000000000003|                         2.16|     Bintulu|          Malaysia|   2.41N|  113.30E|    ALC|\n",
      "|1825-01-01|              26.517|           2.5839999999999996| Butterworth|          Malaysia|   5.63N|  100.09E|    ALC|\n",
      "|1845-01-01|              24.995|                        1.871|      Cainta|       Philippines|  15.27N|  120.83E|    ALC|\n",
      "|1825-01-01|              24.753|           2.1519999999999997|      Ciamis|         Indonesia|   7.23S|  107.84E|    ALC|\n",
      "|1850-01-01|              22.121|           1.5730000000000002|      Dodoma|          Tanzania|   5.63S|   35.52E|    ALC|\n",
      "|1840-01-01|  6.2250000000000005|                        2.112|      Fuling|             China|  29.74N|  107.08E|    ALC|\n",
      "|1841-01-01| -0.3360000000000001|                        2.695|      Fuyang|             China|  32.95N|  115.85E|    ALC|\n",
      "|1856-01-01|              26.901|                        1.359|         Ife|           Nigeria|   7.23N|    4.05E|    ALC|\n",
      "|1796-01-01|  14.427999999999999|                        2.395|  Jhunjhunun|             India|  28.13N|   75.45E|    ALC|\n",
      "|1857-01-01|               25.32|           1.5130000000000001|      Maxixe|        Mozambique|  23.31S|   35.83E|    ALC|\n",
      "|1856-01-01|  26.555000000000003|                        1.308|      Owerri|           Nigeria|   5.63N|    6.46E|    ALC|\n",
      "|1824-01-01|  24.958000000000002|                        1.182|Puerto Plata|Dominican Republic|  20.09N|   69.95W|    ALC|\n",
      "|1832-01-01|              22.288|                        1.587| Santo André|            Brazil|  23.31S|   46.31W|    ALC|\n",
      "|1857-01-01|  18.581000000000003|           1.8119999999999998|      Soweto|      South Africa|  26.52S|   28.66E|    ALC|\n",
      "|1825-01-01|              25.862|                        2.175|    Sukabumi|         Indonesia|   7.23S|  106.22E|    ALC|\n",
      "|1743-11-01|               7.084|           1.8969999999999998|    Toulouse|            France|  44.20N|    2.24E|    ALC|\n",
      "|1835-01-01|              13.086|                        1.942|  Tulancingo|            Mexico|  20.09N|   98.96W|    ALC|\n",
      "|1849-01-01|  26.531999999999996|                        1.416|      Anyama|     Côte D'Ivoire|   5.63N|    4.84W|    ALC|\n",
      "+----------+--------------------+-----------------------------+------------+------------------+--------+---------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Extracting columns Required columns from temperature_dim table to create dimentional table\n",
    "temperature_table = temperature_dim.select([\"City\", \"Country\", \"Latitude\", \"Longitude\",\"AverageTemperature\",\"i94port\"])\n",
    "temperature_table = temperature_dim.withColumnRenamed(\"City\",\"city\")\n",
    "temperature_table.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3490"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "temperature_table.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# writing table to parquet formate\n",
    "temperature_table.write.mode(\"append\").partitionBy(\"i94port\").parquet(\"/results/temperature.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "***Step3***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# path to us-cities-demographic datase\n",
    "path = \"us-cities-demographics.csv\"\n",
    "demographics_table = spark.read.format(\"csv\").option(\"header\",True).option(\"delimiter\",\";\").load(path)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+---------------+-----------------+----------------------+\n",
      "|            city|         state|male_population|female_population|average_household_size|\n",
      "+----------------+--------------+---------------+-----------------+----------------------+\n",
      "|   Silver Spring|      Maryland|          40601|            41862|                   2.6|\n",
      "|          Quincy| Massachusetts|          44129|            49500|                  2.39|\n",
      "|          Hoover|       Alabama|          38040|            46799|                  2.58|\n",
      "|Rancho Cucamonga|    California|          88127|            87105|                  3.18|\n",
      "|          Newark|    New Jersey|         138040|           143873|                  2.73|\n",
      "|          Peoria|      Illinois|          56229|            62432|                   2.4|\n",
      "|        Avondale|       Arizona|          38712|            41971|                  3.18|\n",
      "|     West Covina|    California|          51629|            56860|                  3.56|\n",
      "|        O'Fallon|      Missouri|          41762|            43270|                  2.77|\n",
      "|      High Point|North Carolina|          51751|            58077|                  2.65|\n",
      "|          Folsom|    California|          41051|            35317|                  2.62|\n",
      "|          Folsom|    California|          41051|            35317|                  2.62|\n",
      "|    Philadelphia|  Pennsylvania|         741270|           826172|                  2.61|\n",
      "|         Wichita|        Kansas|         192354|           197601|                  2.56|\n",
      "|         Wichita|        Kansas|         192354|           197601|                  2.56|\n",
      "|      Fort Myers|       Florida|          36850|            37165|                  2.45|\n",
      "|      Pittsburgh|  Pennsylvania|         149690|           154695|                  2.13|\n",
      "|          Laredo|         Texas|         124305|           131484|                  3.66|\n",
      "|        Berkeley|    California|          60142|            60829|                  2.35|\n",
      "|     Santa Clara|    California|          63278|            62938|                  2.75|\n",
      "+----------------+--------------+---------------+-----------------+----------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Extracting required columns to model the dimention table\n",
    "from pyspark.sql.functions import col\n",
    "demographics_dim = demographics_table.select(col(\"City\").alias(\"city\"), col(\"State\").alias(\"state\"), col(\"Male Population\").alias(\"male_population\"), col(\"Female Population\").alias(\"female_population\"), col(\"Average Household Size\").alias(\"average_household_size\"))\n",
    "demographics_dim.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# writing data to results directory in parquet format\n",
    "demographics_dim.write.mode(\"append\").partitionBy(\"city\").parquet(\"/results/demographics.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "***step3***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Creating Fact table from Dimenstion tables which were created above\n",
    "\n",
    "# Create views for each table \n",
    "demographics_dim.createOrReplaceTempView(\"demographics_view\")\n",
    "temperature_table.createOrReplaceTempView(\"temperature_view\")\n",
    "immigration_dim.createOrReplaceTempView(\"immigration_view\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Create immigration_temperature_table by joining immigration dimention table and temperature_table dimention table\n",
    "immigration_temperature_table = immigration_dim.join(temperature_table, on=[\"i94port\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+------+------+-------+-------+-------+-------+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "|i94port| i94yr|i94mon|i94cit|arrdate|i94mode|depdate|i94visa|        dt|  AverageTemperature|AverageTemperatureUncertainty|        city|           Country|Latitude|Longitude|\n",
      "+-------+------+------+------+-------+-------+-------+-------+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1743-11-01|               3.264|                        1.665|   Allentown|     United States|  40.99N|   74.56W|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1779-11-01|0.011999999999999985|                        2.714|      Atyrau|        Kazakhstan|  47.42N|   50.92E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1825-01-01|  26.069000000000003|                         2.16|     Bintulu|          Malaysia|   2.41N|  113.30E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1825-01-01|              26.517|           2.5839999999999996| Butterworth|          Malaysia|   5.63N|  100.09E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1845-01-01|              24.995|                        1.871|      Cainta|       Philippines|  15.27N|  120.83E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1825-01-01|              24.753|           2.1519999999999997|      Ciamis|         Indonesia|   7.23S|  107.84E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1850-01-01|              22.121|           1.5730000000000002|      Dodoma|          Tanzania|   5.63S|   35.52E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1840-01-01|  6.2250000000000005|                        2.112|      Fuling|             China|  29.74N|  107.08E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1841-01-01| -0.3360000000000001|                        2.695|      Fuyang|             China|  32.95N|  115.85E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1856-01-01|              26.901|                        1.359|         Ife|           Nigeria|   7.23N|    4.05E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1796-01-01|  14.427999999999999|                        2.395|  Jhunjhunun|             India|  28.13N|   75.45E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1857-01-01|               25.32|           1.5130000000000001|      Maxixe|        Mozambique|  23.31S|   35.83E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1856-01-01|  26.555000000000003|                        1.308|      Owerri|           Nigeria|   5.63N|    6.46E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1824-01-01|  24.958000000000002|                        1.182|Puerto Plata|Dominican Republic|  20.09N|   69.95W|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1832-01-01|              22.288|                        1.587| Santo André|            Brazil|  23.31S|   46.31W|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1857-01-01|  18.581000000000003|           1.8119999999999998|      Soweto|      South Africa|  26.52S|   28.66E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1825-01-01|              25.862|                        2.175|    Sukabumi|         Indonesia|   7.23S|  106.22E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1743-11-01|               7.084|           1.8969999999999998|    Toulouse|            France|  44.20N|    2.24E|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1835-01-01|              13.086|                        1.942|  Tulancingo|            Mexico|  20.09N|   98.96W|\n",
      "|    ALC|2016.0|   6.0| 103.0|20613.0|    3.0|20626.0|    2.0|1849-01-01|  26.531999999999996|                        1.416|      Anyama|     Côte D'Ivoire|   5.63N|    4.84W|\n",
      "+-------+------+------+------+-------+-------+-------+-------+----------+--------------------+-----------------------------+------------+------------------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "immigration_temperature_table.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- i94port: string (nullable = true)\n",
      " |-- i94yr: double (nullable = true)\n",
      " |-- i94mon: double (nullable = true)\n",
      " |-- i94cit: double (nullable = true)\n",
      " |-- arrdate: double (nullable = true)\n",
      " |-- i94mode: double (nullable = true)\n",
      " |-- depdate: double (nullable = true)\n",
      " |-- i94visa: double (nullable = true)\n",
      " |-- dt: string (nullable = true)\n",
      " |-- AverageTemperature: string (nullable = true)\n",
      " |-- AverageTemperatureUncertainty: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- Country: string (nullable = true)\n",
      " |-- Latitude: string (nullable = true)\n",
      " |-- Longitude: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "immigration_temperature_table.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- year: double (nullable = true)\n",
      " |-- month: double (nullable = true)\n",
      " |-- city_code: double (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      " |-- i94port: string (nullable = true)\n",
      " |-- arrival_date: double (nullable = true)\n",
      " |-- tavel_mode: double (nullable = true)\n",
      " |-- departure_date: double (nullable = true)\n",
      " |-- reason: double (nullable = true)\n",
      " |-- averagetemperature: string (nullable = true)\n",
      " |-- latitude: string (nullable = true)\n",
      " |-- longitude: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# selecting required column for fact table changing column name to readable format\n",
    "fact_table = immigration_temperature_table.select(col(\"i94yr\").alias(\"year\"),\n",
    "                                                col(\"i94mon\").alias(\"month\"),\n",
    "                                                col(\"i94cit\").alias(\"city_code\"), \n",
    "                                                col(\"city\").alias(\"city\"),\n",
    "                                                col(\"Country\").alias(\"country\"),\n",
    "                                                col(\"i94port\").alias(\"i94port\"),\n",
    "                                                col(\"arrdate\").alias(\"arrival_date\"),\n",
    "                                                col(\"i94mon\").alias(\"tavel_mode\"),\n",
    "                                                col(\"depdate\").alias(\"departure_date\"),\n",
    "                                                col(\"i94visa\").alias(\"reason\"),\n",
    "                                                col(\"AverageTemperature\").alias(\"averagetemperature\"),\n",
    "                                                col(\"Latitude\").alias(\"latitude\"),\n",
    "                                                col(\"Longitude\").alias(\"longitude\")\n",
    "                                               )\n",
    "fact_table.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+---------+------------+------------------+-------+------------+----------+--------------+------+--------------------+--------+---------+\n",
      "|  year|month|city_code|        city|           country|i94port|arrival_date|tavel_mode|departure_date|reason|  averagetemperature|latitude|longitude|\n",
      "+------+-----+---------+------------+------------------+-------+------------+----------+--------------+------+--------------------+--------+---------+\n",
      "|2016.0|  6.0|    103.0|   Allentown|     United States|    ALC|     20613.0|       6.0|       20626.0|   2.0|               3.264|  40.99N|   74.56W|\n",
      "|2016.0|  6.0|    103.0|      Atyrau|        Kazakhstan|    ALC|     20613.0|       6.0|       20626.0|   2.0|0.011999999999999985|  47.42N|   50.92E|\n",
      "|2016.0|  6.0|    103.0|     Bintulu|          Malaysia|    ALC|     20613.0|       6.0|       20626.0|   2.0|  26.069000000000003|   2.41N|  113.30E|\n",
      "|2016.0|  6.0|    103.0| Butterworth|          Malaysia|    ALC|     20613.0|       6.0|       20626.0|   2.0|              26.517|   5.63N|  100.09E|\n",
      "|2016.0|  6.0|    103.0|      Cainta|       Philippines|    ALC|     20613.0|       6.0|       20626.0|   2.0|              24.995|  15.27N|  120.83E|\n",
      "|2016.0|  6.0|    103.0|      Ciamis|         Indonesia|    ALC|     20613.0|       6.0|       20626.0|   2.0|              24.753|   7.23S|  107.84E|\n",
      "|2016.0|  6.0|    103.0|      Dodoma|          Tanzania|    ALC|     20613.0|       6.0|       20626.0|   2.0|              22.121|   5.63S|   35.52E|\n",
      "|2016.0|  6.0|    103.0|      Fuling|             China|    ALC|     20613.0|       6.0|       20626.0|   2.0|  6.2250000000000005|  29.74N|  107.08E|\n",
      "|2016.0|  6.0|    103.0|      Fuyang|             China|    ALC|     20613.0|       6.0|       20626.0|   2.0| -0.3360000000000001|  32.95N|  115.85E|\n",
      "|2016.0|  6.0|    103.0|         Ife|           Nigeria|    ALC|     20613.0|       6.0|       20626.0|   2.0|              26.901|   7.23N|    4.05E|\n",
      "|2016.0|  6.0|    103.0|  Jhunjhunun|             India|    ALC|     20613.0|       6.0|       20626.0|   2.0|  14.427999999999999|  28.13N|   75.45E|\n",
      "|2016.0|  6.0|    103.0|      Maxixe|        Mozambique|    ALC|     20613.0|       6.0|       20626.0|   2.0|               25.32|  23.31S|   35.83E|\n",
      "|2016.0|  6.0|    103.0|      Owerri|           Nigeria|    ALC|     20613.0|       6.0|       20626.0|   2.0|  26.555000000000003|   5.63N|    6.46E|\n",
      "|2016.0|  6.0|    103.0|Puerto Plata|Dominican Republic|    ALC|     20613.0|       6.0|       20626.0|   2.0|  24.958000000000002|  20.09N|   69.95W|\n",
      "|2016.0|  6.0|    103.0| Santo André|            Brazil|    ALC|     20613.0|       6.0|       20626.0|   2.0|              22.288|  23.31S|   46.31W|\n",
      "|2016.0|  6.0|    103.0|      Soweto|      South Africa|    ALC|     20613.0|       6.0|       20626.0|   2.0|  18.581000000000003|  26.52S|   28.66E|\n",
      "|2016.0|  6.0|    103.0|    Sukabumi|         Indonesia|    ALC|     20613.0|       6.0|       20626.0|   2.0|              25.862|   7.23S|  106.22E|\n",
      "|2016.0|  6.0|    103.0|    Toulouse|            France|    ALC|     20613.0|       6.0|       20626.0|   2.0|               7.084|  44.20N|    2.24E|\n",
      "|2016.0|  6.0|    103.0|  Tulancingo|            Mexico|    ALC|     20613.0|       6.0|       20626.0|   2.0|              13.086|  20.09N|   98.96W|\n",
      "|2016.0|  6.0|    103.0|      Anyama|     Côte D'Ivoire|    ALC|     20613.0|       6.0|       20626.0|   2.0|  26.531999999999996|   5.63N|    4.84W|\n",
      "+------+-----+---------+------------+------------------+-------+------------+----------+--------------+------+--------------------+--------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fact_table.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2966500"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fact_table.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "success\n"
     ]
    }
   ],
   "source": [
    "fact_table.write.mode(\"append\").partitionBy([\"i94port\",\"city\"]).parquet(\"/results/fact_table.parquet\")\n",
    "print(\"success\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Create another fact table to analyse immigration people in USA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- Create separate  tables for fact_table and demographics tables \n",
    "- Then make sure in fact_table we have only USA cities \n",
    "- Join two tables and make sure we dont have any repeated or null values after joining"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# select required columns from fact_table created above\n",
    "fact_table_immigration = fact_table.select([\"city\", \"i94port\", \"month\", \"reason\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "567"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get all city names in USA and make sure wont have duplicate cities in the list\n",
    "usa_cities_list = [row[\"city\"] for row in demographics_dim.collect()]\n",
    "set_usa_list = set(usa_cities_list)\n",
    "len(set_usa_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# function to verify whether city in usa_cities list or not\n",
    "@udf\n",
    "def get_city(city):\n",
    "    '''\n",
    "    input: city name\n",
    "    \n",
    "    output: if city in set_usa_list return city else return null\n",
    "    '''\n",
    "    if city in set_usa_list:\n",
    "        return city\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Add usa_city column to fact_table_immigration table\n",
    "fact_table_immigration = fact_table_immigration.withColumn(\"usa_city\", get_city(fact_table_immigration.city))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# drop column usa_city \n",
    "fact_table_immigration = fact_table_immigration.drop('usa_city')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Drop duplicate rows in city column\n",
    "fact_table_immigration = fact_table_immigration.dropDuplicates([\"city\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------+-----+------+\n",
      "|        city|i94port|month|reason|\n",
      "+------------+-------+-----+------+\n",
      "|     Antwerp|    ALC|  6.0|   2.0|\n",
      "|    Araruama|    ALC|  6.0|   2.0|\n",
      "|   Bangalore|    ALC|  6.0|   2.0|\n",
      "|       Benxi|    ALC|  6.0|   2.0|\n",
      "|   Cajamarca|    ALC|  6.0|   2.0|\n",
      "|  Cicalengka|    ALC|  6.0|   2.0|\n",
      "|     Copiapo|    ALC|  6.0|   2.0|\n",
      "|     Da Nang|    ALC|  6.0|   2.0|\n",
      "|Dos Hermanas|    ALC|  6.0|   2.0|\n",
      "|    Fengshan|    ALC|  6.0|   2.0|\n",
      "|    Girardot|    ALC|  6.0|   2.0|\n",
      "|     Hanover|    ALC|  6.0|   2.0|\n",
      "|    Horlivka|    ALC|  6.0|   2.0|\n",
      "|    Jaboatão|    ALC|  6.0|   2.0|\n",
      "|   Kisangani|    ALC|  6.0|   2.0|\n",
      "|   Magdeburg|    ALC|  6.0|   2.0|\n",
      "|       Mainz|    ALC|  6.0|   2.0|\n",
      "|    Montería|    ALC|  6.0|   2.0|\n",
      "|     Morelia|    ALC|  6.0|   2.0|\n",
      "|    Namangan|    ALC|  6.0|   2.0|\n",
      "+------------+-------+-----+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# results obtainer\n",
    "fact_table_immigration.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Join fact_table_immigration and demographics_dim table on city \n",
    "fact_table_immigration =  fact_table_immigration.join(demographics_dim, on=[\"city\"])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-------+-----+------+--------------+---------------+-----------------+----------------------+\n",
      "|           city|i94port|month|reason|         state|male_population|female_population|average_household_size|\n",
      "+---------------+-------+-----+------+--------------+---------------+-----------------+----------------------+\n",
      "|      Worcester|    ALC|  6.0|   2.0| Massachusetts|          90951|            93855|                  2.43|\n",
      "|     Charleston|    ALC|  6.0|   2.0|South Carolina|          63956|            71568|                   2.4|\n",
      "|         Corona|    ALC|  6.0|   2.0|    California|          79749|            84493|                  2.97|\n",
      "|    Springfield|    ALC|  6.0|   2.0| Massachusetts|          74744|            79592|                  2.81|\n",
      "|          Tempe|    ALC|  6.0|   2.0|       Arizona|          91350|            84476|                  2.44|\n",
      "|North Las Vegas|    ALC|  6.0|   2.0|        Nevada|         116350|           118443|                  3.31|\n",
      "|       Thornton|    ALC|  6.0|   2.0|      Colorado|          65482|            67977|                  2.97|\n",
      "|        Phoenix|    ALC|  6.0|   2.0|       Arizona|         786833|           776168|                  2.89|\n",
      "|      Hollywood|    ALC|  6.0|   2.0|       Florida|          75358|            74363|                  2.65|\n",
      "| Pembroke Pines|    ALC|  6.0|   2.0|       Florida|          77050|            89574|                  2.94|\n",
      "|       Savannah|    ALC|  6.0|   2.0|       Georgia|          69389|            76295|                  2.57|\n",
      "|     Toms River|    ALC|  6.0|   2.0|    New Jersey|          43864|            43886|                  2.71|\n",
      "|  Coral Springs|    ALC|  6.0|   2.0|       Florida|          63316|            66186|                  3.17|\n",
      "|          Omaha|    ALC|  6.0|   2.0|      Nebraska|         218789|           225098|                  2.47|\n",
      "|      Anchorage|    ALC|  6.0|   2.0|        Alaska|         152945|           145750|                  2.77|\n",
      "|       Paradise|    ALC|  6.0|   2.0|        Nevada|         122784|           114162|                  2.65|\n",
      "|   Fort Collins|    ALC|  6.0|   2.0|      Colorado|          80893|            80288|                  2.45|\n",
      "|        Anaheim|    ALC|  6.0|   2.0|    California|         179603|           171135|                  3.45|\n",
      "|     Greensboro|    ALC|  6.0|   2.0|North Carolina|         132251|           153093|                  2.36|\n",
      "|    Chattanooga|    ALC|  6.0|   2.0|     Tennessee|          83640|            92957|                   2.4|\n",
      "+---------------+-------+-----+------+--------------+---------------+-----------------+----------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Drop duplicate values present in table and remove duplicate values.\n",
    "fact_table_immigration = fact_table_immigration.dropDuplicates(['city'])\n",
    "fact_table_immigration.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "success\n"
     ]
    }
   ],
   "source": [
    "fact_table_immigration.write.mode(\"append\").partitionBy([\"i94port\",\"city\"]).parquet(\"/results/fact_table_immigration.parquet\")\n",
    "print(\"success\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 4.2 Data Quality Checks###"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Explain the data quality checks you'll perform to ensure the pipeline ran as expected. These could include:\n",
    "\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"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Run Quality checks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "def quality_check(df, description):\n",
    "    '''\n",
    "    Input: Spark dataframe, description of Spark datafram\n",
    "    Output: Print outcome of data quality check\n",
    "    '''\n",
    "    \n",
    "    result = df.count()\n",
    "    if result == 0:\n",
    "        print(\"Data quality check failed for {} with zero records\".format(description))\n",
    "    else:\n",
    "        print(\"Data quality check passed for {} with {} records\".format(description, result))\n",
    "    return 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data quality check passed for immigration table with 3508813 records\n",
      "Data quality check passed for temperature table with 3490 records\n",
      "Data quality check passed for demographics table with 2891 records\n",
      "Data quality check passed for fact table with 2966500 records\n",
      "Data quality check passed for fact_table_immigration table with 252 records\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0"
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Perform data quality check\n",
    "quality_check(immigration_dim, \"immigration table\")\n",
    "quality_check(temperature_table, \"temperature table\")\n",
    "quality_check(demographics_dim, \"demographics table\")\n",
    "quality_check(fact_table, \"fact table\")\n",
    "quality_check(fact_table_immigration, \"fact_table_immigration table\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 4.3 Data dictionary \n",
    "Create a data dictionary for your data model. For each field, provide a brief description of what the data is and where it came from. You can include the data dictionary in the notebook or in a separate file."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### fact_table \n",
    "this will contain data by joining i94 immigration to gobaltemperature dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "##### columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- i94yr     AS year \n",
    "- i94mon    AS month\n",
    "- i94cit    AS city_code\n",
    "- city      AS city\n",
    "- Country   AS country\n",
    "- i94port   AS i94port\n",
    "- arrrdate  AS arrival_date\n",
    "- i94visa   AS reason\n",
    "- i94mod    AS travel_mode\n",
    "- depdate   AS departure_date\n",
    "- Average Temperature AS averagetemperature\n",
    "- Latitude  AS latitude\n",
    "- Longitude AS longitude "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### fact_table_immigration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "##### Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- city AS city\n",
    "- i94port AS i94port\n",
    "- i94mon AS month\n",
    "- i94visa AS reason\n",
    "- state AS state\n",
    "- Male Population AS male_population\n",
    "- Female Population AS female_population\n",
    "- Average HouseHold Size AS average_household_size"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### 1.Dimentional Table\n",
    "Contain Data from i94immigration data\n",
    "\n",
    "Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- i94yr = 4 digit year\n",
    "- i94mon = numeric month\n",
    "- i94cit = 3 digit code of origin city\n",
    "- i94port = 3 character code of destination city\n",
    "- arrdate = arrival date\n",
    "- i94mode = 1 digit travel code\n",
    "- depdate = departure date\n",
    "- i94visa = reason for immigration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 2.Dimentional Table\n",
    "Contain data from global temperature data\n",
    "\n",
    "Columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- 94port = 3 character code of destination city (mapped from cleaned up immigration data)\n",
    "- AverageTemperature = average temperature\n",
    "- City = city name\n",
    "- Country = country name\n",
    "- Latitude= latitude\n",
    "- Longitude = longitude"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### 3.Dimentional Table\n",
    "Contain data from us-cities-demograhics dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "Columns\n",
    "- "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "- city  AS city\n",
    "- state AS state\n",
    "- Male Population AS male_population\n",
    "- Female Population AS female_population\n",
    "- Average HouseHold Size AS average_household_size"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "## Step 5: Complete Project Write Up\n",
    "1 Clearly state the rationale for the choice of tools and technologies for the project.\n",
    "\n",
    "- we used Spark for this project since it can easily handle multiple file formats (json,SAS, csv,text, etc) that contain large i.e 1M amounts of data. Its mainly because ***Schema-on-Read***(no database storage required) which is main advantage over datawarehouse.Spark is know for **Lazy Evaluation** means whenever you want data,then it run quiries and take data from disk to memory. Spark SQL was used to process the input files into dataframes and manipulated via standard SQL join operations to create the tables.\n",
    "\n",
    "2 Propose how often the data should be updated and why.\n",
    "\n",
    "- Data is updated for every months. we can analyse data by pullin for every months.\n",
    "\n",
    "3 Write a description of how you would approach the problem differently under the following scenarios:\n",
    " * The data was increased by 100x.\n",
    " \n",
    " - AWS provides AmazonRedshift using which we can analyze millions of data workloads easily\n",
    " \n",
    "4 The data populates a dashboard that must be updated on a daily basis by 7am every day.\n",
    " \n",
    " - Using Airflow DAG we can trigger the task for every 7am. if task fails sending emails can be done.\n",
    " \n",
    " - DataQualitycheck pulgins can be included in our task for every load data we analyse.\n",
    " \n",
    "5 The database needed to be accessed by 100+ people.\n",
    " \n",
    " - Using Redshift we can solve this problem, it has autoscaling capabilities and good read performance\n",
    " "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}