smart_advertisement_capstone.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Smart Advertisement Service\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Business scenario\n",
    "\n",
    "An advertising consultant startup in US, which focuses on consulting range of business on effective advertisement, wants to adapt the cutting edge technology in order to enhance their quality of services. Lately conducted research on company has found out that around 14 percent of the national population are immigrant, [source](https://www.americanimmigrationcouncil.org/research/immigrants-in-the-united-states) . Thus the company has decided to build a data warehouse which will used for analytics for better consultation on advertisement. Not limited to that, the warehouse will also be used as a brain for backend services that will possibly automate the advertisement of certain category."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Structure of the Project\n",
    "\n",
    "The project follows the follow steps:\n",
    "* Step 1: Scope the Project and Gather Data\n",
    "* Step 2: Explore and Assess the Data\n",
    "* Step 3: Define the Data Model\n",
    "* Step 4: Run ETL to Model the Data\n",
    "* Step 5: Complete Project Write Up"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 1: Scope the Project and Gather Data\n",
    "\n",
    "#### Scope\n",
    "In this project I will create a cloud data warehouse that will support answering question through analytics tables and dashboards. Later, the warehouse could be used as a backend for developing automatic advertisement services.\n",
    "\n",
    "The following steps will be carried out:\n",
    "- Transform the data and store in s3 in parquet format\n",
    "- Load the data to redshift for analytics\n",
    "- Use airflow to monitor pipeline\n",
    "- Use SQL client tool to the redshift database for analytics\n",
    "- Later, api will be developed using lambda function and API gateway for dashboard and automatic advertisement\n",
    "\n",
    "#### The data\n",
    "\n",
    "- I94 immigration data that comes from the US National Tourism and Trade Office, [source](https://www.trade.gov/national-travel-and-tourism-office).\n",
    "- U.S City demographic data that comes from Opensoft, [source](https://www.kaggle.com/berkeleyearth/climate-change-earth-surface-temperature-data).\n",
    "- Airports data from Datahub, [source](https://datahub.io/core/airport-codes#data)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import os\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import types as T\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import IntegerType\n",
    "from datetime import datetime, timedelta\n",
    "import json\n",
    "from pyspark.sql.functions import desc, monotonically_increasing_id, udf, to_date, from_unixtime, trim, col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "21/12/08 01:23:06 WARN Utils: Your hostname, Yugeshs-MacBook-Pro.local resolves to a loopback address: 127.0.0.1; using 192.168.0.18 instead (on interface en0)\n",
      "21/12/08 01:23:06 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/Users/yugesh/opt/anaconda3/envs/airflow/lib/python3.8/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /Users/yugesh/.ivy2/cache\n",
      "The jars for the packages stored in: /Users/yugesh/.ivy2/jars\n",
      "org.apache.hadoop#hadoop-aws added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-5e90d866-2492-43be-b2cf-dede2821d5a4;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.hadoop#hadoop-aws;2.7.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-common;2.7.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-annotations;2.7.0 in central\n",
      "\tfound com.google.guava#guava;11.0.2 in central\n",
      "\tfound com.google.code.findbugs#jsr305;3.0.0 in central\n",
      "\tfound commons-cli#commons-cli;1.2 in central\n",
      "\tfound org.apache.commons#commons-math3;3.1.1 in central\n",
      "\tfound xmlenc#xmlenc;0.52 in central\n",
      "\tfound commons-httpclient#commons-httpclient;3.1 in central\n",
      "\tfound commons-logging#commons-logging;1.1.3 in central\n",
      "\tfound commons-codec#commons-codec;1.4 in central\n",
      "\tfound commons-io#commons-io;2.4 in central\n",
      "\tfound commons-net#commons-net;3.1 in central\n",
      "\tfound commons-collections#commons-collections;3.2.1 in central\n",
      "\tfound javax.servlet#servlet-api;2.5 in central\n",
      "\tfound org.mortbay.jetty#jetty;6.1.26 in central\n",
      "\tfound org.mortbay.jetty#jetty-util;6.1.26 in central\n",
      "\tfound com.sun.jersey#jersey-core;1.9 in central\n",
      "\tfound com.sun.jersey#jersey-json;1.9 in central\n",
      "\tfound org.codehaus.jettison#jettison;1.1 in central\n",
      "\tfound com.sun.xml.bind#jaxb-impl;2.2.3-1 in central\n",
      "\tfound javax.xml.bind#jaxb-api;2.2.2 in central\n",
      "\tfound javax.xml.stream#stax-api;1.0-2 in central\n",
      "\tfound javax.activation#activation;1.1 in central\n",
      "\tfound org.codehaus.jackson#jackson-core-asl;1.9.13 in central\n",
      "\tfound org.codehaus.jackson#jackson-mapper-asl;1.9.13 in central\n",
      "\tfound org.codehaus.jackson#jackson-jaxrs;1.9.13 in central\n",
      "\tfound org.codehaus.jackson#jackson-xc;1.9.13 in central\n",
      "\tfound com.sun.jersey#jersey-server;1.9 in central\n",
      "\tfound asm#asm;3.2 in central\n",
      "\tfound log4j#log4j;1.2.17 in central\n",
      "\tfound net.java.dev.jets3t#jets3t;0.9.0 in central\n",
      "\tfound org.apache.httpcomponents#httpclient;4.2.5 in central\n",
      "\tfound org.apache.httpcomponents#httpcore;4.2.5 in central\n",
      "\tfound com.jamesmurty.utils#java-xmlbuilder;0.4 in central\n",
      "\tfound commons-lang#commons-lang;2.6 in central\n",
      "\tfound commons-configuration#commons-configuration;1.6 in central\n",
      "\tfound commons-digester#commons-digester;1.8 in central\n",
      "\tfound commons-beanutils#commons-beanutils;1.7.0 in central\n",
      "\tfound commons-beanutils#commons-beanutils-core;1.8.0 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.10 in central\n",
      "\tfound org.apache.avro#avro;1.7.4 in central\n",
      "\tfound com.thoughtworks.paranamer#paranamer;2.3 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.0.4.1 in central\n",
      "\tfound org.apache.commons#commons-compress;1.4.1 in central\n",
      "\tfound org.tukaani#xz;1.0 in central\n",
      "\tfound com.google.protobuf#protobuf-java;2.5.0 in central\n",
      "\tfound com.google.code.gson#gson;2.2.4 in central\n",
      "\tfound org.apache.hadoop#hadoop-auth;2.7.0 in central\n",
      "\tfound org.apache.directory.server#apacheds-kerberos-codec;2.0.0-M15 in central\n",
      "\tfound org.apache.directory.server#apacheds-i18n;2.0.0-M15 in central\n",
      "\tfound org.apache.directory.api#api-asn1-api;1.0.0-M20 in central\n",
      "\tfound org.apache.directory.api#api-util;1.0.0-M20 in central\n",
      "\tfound org.apache.zookeeper#zookeeper;3.4.6 in central\n",
      "\tfound org.slf4j#slf4j-log4j12;1.7.10 in central\n",
      "\tfound io.netty#netty;3.6.2.Final in central\n",
      "\tfound org.apache.curator#curator-framework;2.7.1 in central\n",
      "\tfound org.apache.curator#curator-client;2.7.1 in central\n",
      "\tfound com.jcraft#jsch;0.1.42 in central\n",
      "\tfound org.apache.curator#curator-recipes;2.7.1 in central\n",
      "\tfound org.apache.htrace#htrace-core;3.1.0-incubating in central\n",
      "\tfound javax.servlet.jsp#jsp-api;2.1 in central\n",
      "\tfound jline#jline;0.9.94 in central\n",
      "\tfound com.fasterxml.jackson.core#jackson-databind;2.2.3 in central\n",
      "\tfound com.fasterxml.jackson.core#jackson-annotations;2.2.3 in central\n",
      "\tfound com.fasterxml.jackson.core#jackson-core;2.2.3 in central\n",
      "\tfound com.amazonaws#aws-java-sdk;1.7.4 in central\n",
      "\tfound joda-time#joda-time;2.10.13 in central\n",
      "\t[2.10.13] joda-time#joda-time;[2.2,)\n",
      ":: resolution report :: resolve 7875ms :: artifacts dl 201ms\n",
      "\t:: modules in use:\n",
      "\tasm#asm;3.2 from central in [default]\n",
      "\tcom.amazonaws#aws-java-sdk;1.7.4 from central in [default]\n",
      "\tcom.fasterxml.jackson.core#jackson-annotations;2.2.3 from central in [default]\n",
      "\tcom.fasterxml.jackson.core#jackson-core;2.2.3 from central in [default]\n",
      "\tcom.fasterxml.jackson.core#jackson-databind;2.2.3 from central in [default]\n",
      "\tcom.google.code.findbugs#jsr305;3.0.0 from central in [default]\n",
      "\tcom.google.code.gson#gson;2.2.4 from central in [default]\n",
      "\tcom.google.guava#guava;11.0.2 from central in [default]\n",
      "\tcom.google.protobuf#protobuf-java;2.5.0 from central in [default]\n",
      "\tcom.jamesmurty.utils#java-xmlbuilder;0.4 from central in [default]\n",
      "\tcom.jcraft#jsch;0.1.42 from central in [default]\n",
      "\tcom.sun.jersey#jersey-core;1.9 from central in [default]\n",
      "\tcom.sun.jersey#jersey-json;1.9 from central in [default]\n",
      "\tcom.sun.jersey#jersey-server;1.9 from central in [default]\n",
      "\tcom.sun.xml.bind#jaxb-impl;2.2.3-1 from central in [default]\n",
      "\tcom.thoughtworks.paranamer#paranamer;2.3 from central in [default]\n",
      "\tcommons-beanutils#commons-beanutils;1.7.0 from central in [default]\n",
      "\tcommons-beanutils#commons-beanutils-core;1.8.0 from central in [default]\n",
      "\tcommons-cli#commons-cli;1.2 from central in [default]\n",
      "\tcommons-codec#commons-codec;1.4 from central in [default]\n",
      "\tcommons-collections#commons-collections;3.2.1 from central in [default]\n",
      "\tcommons-configuration#commons-configuration;1.6 from central in [default]\n",
      "\tcommons-digester#commons-digester;1.8 from central in [default]\n",
      "\tcommons-httpclient#commons-httpclient;3.1 from central in [default]\n",
      "\tcommons-io#commons-io;2.4 from central in [default]\n",
      "\tcommons-lang#commons-lang;2.6 from central in [default]\n",
      "\tcommons-logging#commons-logging;1.1.3 from central in [default]\n",
      "\tcommons-net#commons-net;3.1 from central in [default]\n",
      "\tio.netty#netty;3.6.2.Final from central in [default]\n",
      "\tjavax.activation#activation;1.1 from central in [default]\n",
      "\tjavax.servlet#servlet-api;2.5 from central in [default]\n",
      "\tjavax.servlet.jsp#jsp-api;2.1 from central in [default]\n",
      "\tjavax.xml.bind#jaxb-api;2.2.2 from central in [default]\n",
      "\tjavax.xml.stream#stax-api;1.0-2 from central in [default]\n",
      "\tjline#jline;0.9.94 from central in [default]\n",
      "\tjoda-time#joda-time;2.10.13 from central in [default]\n",
      "\tlog4j#log4j;1.2.17 from central in [default]\n",
      "\tnet.java.dev.jets3t#jets3t;0.9.0 from central in [default]\n",
      "\torg.apache.avro#avro;1.7.4 from central in [default]\n",
      "\torg.apache.commons#commons-compress;1.4.1 from central in [default]\n",
      "\torg.apache.commons#commons-math3;3.1.1 from central in [default]\n",
      "\torg.apache.curator#curator-client;2.7.1 from central in [default]\n",
      "\torg.apache.curator#curator-framework;2.7.1 from central in [default]\n",
      "\torg.apache.curator#curator-recipes;2.7.1 from central in [default]\n",
      "\torg.apache.directory.api#api-asn1-api;1.0.0-M20 from central in [default]\n",
      "\torg.apache.directory.api#api-util;1.0.0-M20 from central in [default]\n",
      "\torg.apache.directory.server#apacheds-i18n;2.0.0-M15 from central in [default]\n",
      "\torg.apache.directory.server#apacheds-kerberos-codec;2.0.0-M15 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-annotations;2.7.0 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-auth;2.7.0 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-aws;2.7.0 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-common;2.7.0 from central in [default]\n",
      "\torg.apache.htrace#htrace-core;3.1.0-incubating from central in [default]\n",
      "\torg.apache.httpcomponents#httpclient;4.2.5 from central in [default]\n",
      "\torg.apache.httpcomponents#httpcore;4.2.5 from central in [default]\n",
      "\torg.apache.zookeeper#zookeeper;3.4.6 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-core-asl;1.9.13 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-jaxrs;1.9.13 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-mapper-asl;1.9.13 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-xc;1.9.13 from central in [default]\n",
      "\torg.codehaus.jettison#jettison;1.1 from central in [default]\n",
      "\torg.mortbay.jetty#jetty;6.1.26 from central in [default]\n",
      "\torg.mortbay.jetty#jetty-util;6.1.26 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.10 from central in [default]\n",
      "\torg.slf4j#slf4j-log4j12;1.7.10 from central in [default]\n",
      "\torg.tukaani#xz;1.0 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.0.4.1 from central in [default]\n",
      "\txmlenc#xmlenc;0.52 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   68  |   1   |   0   |   0   ||   68  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-5e90d866-2492-43be-b2cf-dede2821d5a4\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 68 already retrieved (0kB/69ms)\n",
      "21/12/08 01:23:15 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "# Create spark session\n",
    "spark = SparkSession.builder.config(\n",
    "    \"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0\"\n",
    ").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    }
   ],
   "source": [
    "# Read in the data here\n",
    "# partial\n",
    "# immigration_df = spark.read.parquet(\"./data/sas_data/part-00000-b9542815-7a8d-45fc-9c67-c9c5007ad0d4-c000.snappy.parquet\")\n",
    "# all\n",
    "immigration_df = spark.read.parquet(\"./data/sas_data/\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "data": {
      "text/plain": [
       "3096313"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "immigration_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 2: Explore and Assess the Data\n",
    "#### Explore the Data \n",
    "Identify data quality issues, like missing values, duplicate data, etc.\n",
    "\n",
    "#### Cleaning Steps\n",
    "Document steps necessary to clean the data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exploring immigration data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cicid: double (nullable = true)\n",
      " |-- i94yr: double (nullable = true)\n",
      " |-- i94mon: double (nullable = true)\n",
      " |-- i94cit: double (nullable = true)\n",
      " |-- i94res: double (nullable = true)\n",
      " |-- i94port: string (nullable = true)\n",
      " |-- arrdate: double (nullable = true)\n",
      " |-- i94mode: double (nullable = true)\n",
      " |-- i94addr: string (nullable = true)\n",
      " |-- depdate: double (nullable = true)\n",
      " |-- i94bir: double (nullable = true)\n",
      " |-- i94visa: double (nullable = true)\n",
      " |-- count: double (nullable = true)\n",
      " |-- dtadfile: string (nullable = true)\n",
      " |-- visapost: string (nullable = true)\n",
      " |-- occup: string (nullable = true)\n",
      " |-- entdepa: string (nullable = true)\n",
      " |-- entdepd: string (nullable = true)\n",
      " |-- entdepu: string (nullable = true)\n",
      " |-- matflag: string (nullable = true)\n",
      " |-- biryear: double (nullable = true)\n",
      " |-- dtaddto: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- insnum: string (nullable = true)\n",
      " |-- airline: string (nullable = true)\n",
      " |-- admnum: double (nullable = true)\n",
      " |-- fltno: string (nullable = true)\n",
      " |-- visatype: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "immigration_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Convert double to integer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+------+-------+-------+-------+------+-------+\n",
      "|  cicid|i94cit|i94res|arrdate|i94mode|depdate|i94bir|i94visa|\n",
      "+-------+------+------+-------+-------+-------+------+-------+\n",
      "|5748517|   245|   438|  20574|      1|  20582|    40|      1|\n",
      "+-------+------+------+-------+-------+-------+------+-------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "int_cols = [\"cicid\", \"i94cit\", \"i94res\", \"arrdate\", \"i94mode\", \"depdate\", \"i94bir\", \"i94visa\"]\n",
    "# Additional options \"dtadfile\", \"daddto\"\n",
    "\n",
    "for col_name in int_cols:\n",
    "    immigration_df = immigration_df.withColumn(col_name, immigration_df[col_name].cast(IntegerType()))\n",
    "\n",
    "immigration_df.select(int_cols).show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Remove duplicate values\n",
    "\n",
    "- Drop duplicate with original ccid, however unable to remove duplicate because of ccid.\n",
    "- Drop duplicate without ccid and recreate ccid able to remove ()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of deleted duplicate values 0\n",
      "Total number of availabel records 3096313\n"
     ]
    }
   ],
   "source": [
    "# Dropping duplicate with ccid\n",
    "with_duplicate = immigration_df.count()\n",
    "\n",
    "immigration_df.dropDuplicates()\n",
    "\n",
    "without_duplicate = immigration_df.count()\n",
    "print(f\"Total number of deleted duplicate values {with_duplicate - without_duplicate}\")\n",
    "print(f\"Total number of availabel records {immigration_df.count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "21/12/08 01:23:36 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of deleted duplicate values 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of available records 3096313\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    }
   ],
   "source": [
    "# Dropping duplicate by excluding ccid\n",
    "\n",
    "with_duplicate = immigration_df.count()\n",
    "\n",
    "immigration_df = immigration_df.drop(\"cicid\")\n",
    "immigration_df = immigration_df.dropDuplicates()\n",
    "\n",
    "immigration_df = immigration_df.withColumn(\"cicid\", monotonically_increasing_id())\n",
    "\n",
    "without_duplicate = immigration_df.count()\n",
    "print(f\"Total number of deleted duplicate values {with_duplicate - without_duplicate}\")\n",
    "print(f\"Total number of available records {immigration_df.count()}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Assign 0 to all null in integer column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "immigration_df = immigration_df.fillna(0, int_cols)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Assign real values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def assign_value(col_name, key):\n",
    "    ROOT = \"./data/lookup/\"\n",
    "    key = str(key)\n",
    "    col_name = col_name.lower()\n",
    "\n",
    "    filepath = os.path.join(ROOT, f\"{col_name}.json\")\n",
    "    with open(filepath, \"r\") as f:\n",
    "        data = json.load(f)\n",
    "\n",
    "    if col_name == \"i94port\":\n",
    "        return data[key].split(\",\")[0] if key in data else None\n",
    "\n",
    "    return data[key] if key in data else None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve transporation mode using i94mode\n",
    "get_mode_udf = udf(lambda x: assign_value(key=x, col_name=\"i94mode\"), T.StringType())\n",
    "immigration_df = immigration_df.withColumn(\"transportation_mode\", get_mode_udf(immigration_df.i94mode))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve arrived city\n",
    "get_city_udf = udf(lambda x: assign_value(key=x, col_name=\"i94port\"), T.StringType())\n",
    "immigration_df = immigration_df.withColumn(\"arrived_city\", get_city_udf(immigration_df.i94port))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve arrived state\n",
    "get_state_udf = udf(lambda x: assign_value(key=x, col_name=\"i94addr\"), T.StringType())\n",
    "immigration_df = immigration_df.withColumn(\"us_address\", get_state_udf(immigration_df.i94addr))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve origin city and travelled from using i94CIT and i94res\n",
    "get_origin_udf = udf(lambda x: assign_value(key=x, col_name=\"i94cit\"), T.StringType())\n",
    "immigration_df = immigration_df.withColumn(\"origin_city\", get_origin_udf(immigration_df.i94cit)).withColumn(\"traveled_from\", get_origin_udf(immigration_df.i94res))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrive i94visa with value\n",
    "get_visa_udf = udf(lambda x: assign_value(key=x, col_name=\"i94visa\"), T.StringType())\n",
    "immigration_df = immigration_df.withColumn(\"visa_status\", get_visa_udf(immigration_df.i94visa))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exclude unused columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No of columns before removing: 34\n",
      "No of columns after removing: 15\n"
     ]
    }
   ],
   "source": [
    "unused_cols = [ \"i94yr\",\"i94mon\",\"count\", \"fltno\",  \"insnum\", \"entdepd\", \"biryear\", \"dtadfile\", \"biryear\", \"visapost\", \"entdepu\", \"admnum\", \"i94cit\", \"i94res\", \"i94port\", \"i94addr\", \"i94mode\", \"i94visa\", \"entdepa\", \"dtaddto\"]\n",
    "print(f\"No of columns before removing: {len(immigration_df.columns)}\")\n",
    "immigration_df = immigration_df.drop(*unused_cols)\n",
    "print(f\"No of columns after removing: {len(immigration_df.columns)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- arrdate: integer (nullable = true)\n",
      " |-- depdate: integer (nullable = true)\n",
      " |-- i94bir: integer (nullable = true)\n",
      " |-- occup: string (nullable = true)\n",
      " |-- matflag: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- airline: string (nullable = true)\n",
      " |-- visatype: string (nullable = true)\n",
      " |-- cicid: long (nullable = false)\n",
      " |-- transportation_mode: string (nullable = true)\n",
      " |-- arrived_city: string (nullable = true)\n",
      " |-- us_address: string (nullable = true)\n",
      " |-- origin_city: string (nullable = true)\n",
      " |-- traveled_from: string (nullable = true)\n",
      " |-- visa_status: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "immigration_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Rename columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Rename columns\n",
    "immigration_df = immigration_df.withColumnRenamed(\"arrdate\", \"arrival_date\").withColumnRenamed(\"depdate\", \"departure_date\").withColumnRenamed(\"i94bir\", \"age\").withColumnRenamed(\"occup\", \"occupation\").withColumnRenamed(\"matflag\", \"matched_flag\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Order the columns in proper sequence"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "immigration_df = immigration_df.select([\"cicid\", \"origin_city\", \"traveled_from\", \"arrived_city\", \"us_address\", \"arrival_date\", \"departure_date\", \"transportation_mode\", \"age\", \"gender\",\n",
    "\"visa_status\", \"occupation\", \"airline\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Convert sas date format to \"YYYY_MM_DD\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convert_datetime(x):\n",
    "    try:\n",
    "        start = datetime(1960, 1, 1)\n",
    "        return start + timedelta(days=int(x))\n",
    "    except:\n",
    "        return None\n",
    "udf_datetime_from_sas = udf(lambda x: convert_datetime(x), T.DateType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "immigration_df = immigration_df.withColumn(\n",
    "    \"arrival_date\", udf_datetime_from_sas(immigration_df.arrival_date)\n",
    ").withColumn(\"departure_date\", udf_datetime_from_sas(immigration_df.departure_date))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exploring us cities demographics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "filepath = \"./data/us_cities_demographics.csv\"\n",
    "demographic_df = spark.read.csv(filepath,inferSchema=True, header=True, sep=';')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- Median Age: double (nullable = true)\n",
      " |-- Male Population: integer (nullable = true)\n",
      " |-- Female Population: integer (nullable = true)\n",
      " |-- Total Population: integer (nullable = true)\n",
      " |-- Number of Veterans: integer (nullable = true)\n",
      " |-- Foreign-born: integer (nullable = true)\n",
      " |-- Average Household Size: double (nullable = true)\n",
      " |-- State Code: string (nullable = true)\n",
      " |-- Race: string (nullable = true)\n",
      " |-- Count: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "demographic_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Removing duplicates: Pivoting column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|                race|\n",
      "+--------------------+\n",
      "|Black or African-...|\n",
      "|  Hispanic or Latino|\n",
      "|               White|\n",
      "|               Asian|\n",
      "|American Indian a...|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Finding types of races\n",
    "demographic_df.select(\"race\").distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pivot column Race to different columns\n",
    "pivot_cols = [\"City\", \"State\"]\n",
    "pivot_df = demographic_df.groupBy(pivot_cols).pivot(\"Race\").sum(\"Count\")\n",
    "\n",
    "# Joining the pivot \n",
    "demographic_df = demographic_df.join(other=pivot_df, on=pivot_cols)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- Median Age: double (nullable = true)\n",
      " |-- Male Population: integer (nullable = true)\n",
      " |-- Female Population: integer (nullable = true)\n",
      " |-- Total Population: integer (nullable = true)\n",
      " |-- Number of Veterans: integer (nullable = true)\n",
      " |-- Foreign-born: integer (nullable = true)\n",
      " |-- Average Household Size: double (nullable = true)\n",
      " |-- State Code: string (nullable = true)\n",
      " |-- Race: string (nullable = true)\n",
      " |-- Count: integer (nullable = true)\n",
      " |-- American Indian and Alaska Native: long (nullable = true)\n",
      " |-- Asian: long (nullable = true)\n",
      " |-- Black or African-American: long (nullable = true)\n",
      " |-- Hispanic or Latino: long (nullable = true)\n",
      " |-- White: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "demographic_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Remove unwanted columns from the schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+---------------+-----------------+----------------+---------------------------------+-----+-------------------------+------------------+------+\n",
      "|      City|State|Male Population|Female Population|Total Population|American Indian and Alaska Native|Asian|Black or African-American|Hispanic or Latino| White|\n",
      "+----------+-----+---------------+-----------------+----------------+---------------------------------+-----+-------------------------+------------------+------+\n",
      "|Cincinnati| Ohio|         143654|           154883|          298537|                             3362| 7633|                   133430|              9121|162245|\n",
      "+----------+-----+---------------+-----------------+----------------+---------------------------------+-----+-------------------------+------------------+------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    }
   ],
   "source": [
    "del_cols = [\"median age\", \"Number of Veterans\", \"foreign-born\", \"Average Household Size\", \"State Code\", \"race\", \"count\"]\n",
    "demographic_df = demographic_df.drop(*del_cols)\n",
    "demographic_df.show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Dropping duplicates "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of deleted duplicate values 2295\n",
      "Total number of availabel records 596\n"
     ]
    }
   ],
   "source": [
    "# Dropping duplicate by excluding ccid\n",
    "\n",
    "with_duplicate = demographic_df.count()\n",
    "\n",
    "demographic_df = demographic_df.dropDuplicates()\n",
    "\n",
    "without_duplicate = demographic_df.count()\n",
    "print(f\"Total number of deleted duplicate values {with_duplicate - without_duplicate}\")\n",
    "print(f\"Total number of availabel records {demographic_df.count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- City: string (nullable = true)\n",
      " |-- State: string (nullable = true)\n",
      " |-- Male Population: integer (nullable = true)\n",
      " |-- Female Population: integer (nullable = true)\n",
      " |-- Total Population: integer (nullable = true)\n",
      " |-- American Indian and Alaska Native: long (nullable = true)\n",
      " |-- Asian: long (nullable = true)\n",
      " |-- Black or African-American: long (nullable = true)\n",
      " |-- Hispanic or Latino: long (nullable = true)\n",
      " |-- White: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "demographic_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Convert column names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "demographic_df = demographic_df.toDF(\"city\", \"state\", \"male_population\", \"female_population\", \"total_population\", \"american_indian_alaska_native\", \"asian\", \"black_african_american\", \"hispanic_latino\", \"white\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- city: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- male_population: integer (nullable = true)\n",
      " |-- female_population: integer (nullable = true)\n",
      " |-- total_population: integer (nullable = true)\n",
      " |-- american_indian_alaska_native: long (nullable = true)\n",
      " |-- asian: long (nullable = true)\n",
      " |-- black_african_american: long (nullable = true)\n",
      " |-- hispanic_latino: long (nullable = true)\n",
      " |-- white: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "demographic_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Filling null with 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_cols = [\"male_population\", \"female_population\", \"total_population\",  \"american_indian_alaska_native\", \"asian\", \"black_african_american\", \"hispanic_latino\", \"white\"]\n",
    "demographic_df = demographic_df.fillna(0, num_cols)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Add id column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- city: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- male_population: integer (nullable = true)\n",
      " |-- female_population: integer (nullable = true)\n",
      " |-- total_population: integer (nullable = true)\n",
      " |-- american_indian_alaska_native: long (nullable = true)\n",
      " |-- asian: long (nullable = true)\n",
      " |-- black_african_american: long (nullable = true)\n",
      " |-- hispanic_latino: long (nullable = true)\n",
      " |-- white: long (nullable = true)\n",
      " |-- demographic_id: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "demographic_df = demographic_df.withColumn(\"demographic_id\", monotonically_increasing_id())\n",
    "demographic_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Reordering the columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "demographic_df = demographic_df.select([\"demographic_id\", \"city\", \"state\", \"american_indian_alaska_native\",\n",
    "\"asian\",\n",
    "\"black_african_american\",\n",
    "\"hispanic_latino\", \"white\", \"male_population\", \"female_population\", \"total_population\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- demographic_id: long (nullable = false)\n",
      " |-- city: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- american_indian_alaska_native: long (nullable = true)\n",
      " |-- asian: long (nullable = true)\n",
      " |-- black_african_american: long (nullable = true)\n",
      " |-- hispanic_latino: long (nullable = true)\n",
      " |-- white: long (nullable = true)\n",
      " |-- male_population: integer (nullable = true)\n",
      " |-- female_population: integer (nullable = true)\n",
      " |-- total_population: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "demographic_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exploring airport data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ident   : string (nullable = true)\n",
      " |-- type           : string (nullable = true)\n",
      " |-- name                                                                                                                               : string (nullable = true)\n",
      " |-- elevation_ft : string (nullable = true)\n",
      " |-- continent : string (nullable = true)\n",
      " |-- iso_country : string (nullable = true)\n",
      " |-- iso_region : string (nullable = true)\n",
      " |-- municipality                                                   : string (nullable = true)\n",
      " |-- gps_code : string (nullable = true)\n",
      " |-- iata_code : string (nullable = true)\n",
      " |-- local_code : string (nullable = true)\n",
      " |-- coordinates: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# reading csv file\n",
    "airport_csv_path = \"./data/airport_codes.csv\"\n",
    "airport_df = spark.read.csv(airport_csv_path, header=True, sep=',', ignoreTrailingWhiteSpace=False)\n",
    "\n",
    "airport_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Removing whitespace in column name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------------+--------------------+-------------+----------+------------+-----------+--------------------+---------+----------+-----------+--------------------+\n",
      "|   ident|           type|                name| elevation_ft| continent| iso_country| iso_region|        municipality| gps_code| iata_code| local_code|         coordinates|\n",
      "+--------+---------------+--------------------+-------------+----------+------------+-----------+--------------------+---------+----------+-----------+--------------------+\n",
      "|00A     |heliport       |Total Rf Heliport...|11           |NA        |US          |US-PA      |Bensalem         ...|00A      |          |00A        |-74.9336013793945...|\n",
      "+--------+---------------+--------------------+-------------+----------+------------+-----------+--------------------+---------+----------+-----------+--------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airport_df = airport_df.select([F.col(col).alias(col.replace(' ', '')) for col in airport_df.columns])\n",
    "airport_df.show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Excluding airport outside of US"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "22757"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "airport_df = airport_df.filter(airport_df.iso_country.like(\"%US%\"))\n",
    "airport_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Add state_code column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "get_state_code_udf = udf(lambda x :  x if x is None else x.split(\"-\")[1])\n",
    "airport_df = airport_df.withColumn(\"state_code\", get_state_code_udf(\"iso_region\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Retrieve state_name from state code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "airport_df = airport_df.withColumn(\"state\", get_state_udf(airport_df.state_code))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Remove unwanted columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ident: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- municipality: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airport_delete_cols = [\"elevation_ft\", \"continent\", \"iso_country\", \"iso_region\" , \"gps_code\", \"iata_code\", \"local_code\", \"coordinates\", \"state_code\"]\n",
    "airport_df = airport_df.drop(*airport_delete_cols)\n",
    "airport_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Drop duplicates"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of deleted values 120\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of available records 22637\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    }
   ],
   "source": [
    "# Dropping duplicate by excluding ident\n",
    "\n",
    "with_duplicate = airport_df.count()\n",
    "\n",
    "airport_df = airport_df.drop(\"ident\")\n",
    "airport_df = airport_df.dropDuplicates()\n",
    "\n",
    "airport_df = airport_df.withColumn(\"airport_id\", monotonically_increasing_id())\n",
    "\n",
    "without_duplicate = airport_df.count()\n",
    "print(f\"Total number of deleted values {with_duplicate - without_duplicate}\")\n",
    "print(f\"Total number of available records {airport_df.count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- type: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- municipality: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- airport_id: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airport_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Rename munacipality to city"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "airport_df = airport_df.withColumnRenamed(\"municipality\", \"city\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Reordering columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- airport_id: long (nullable = false)\n",
      " |-- name: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airport_df = airport_df.select([\"airport_id\", \"name\", \"city\", \"state\", \"type\"])\n",
    "airport_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Remove space from columns values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "remove_whitespace_udf = udf(lambda x: str(x).strip())\n",
    "airport_df = (\n",
    "    airport_df.withColumn(\n",
    "        \"name\",\n",
    "        remove_whitespace_udf(airport_df.name)\n",
    "        if airport_df.name.isNotNull\n",
    "        else airport_df.name,\n",
    "    )\n",
    "    .withColumn(\n",
    "        \"city\",\n",
    "        remove_whitespace_udf(airport_df.city)\n",
    "        if airport_df.city.isNotNull\n",
    "        else airport_df.city,\n",
    "    )\n",
    "    .withColumn(\n",
    "        \"state\",\n",
    "        remove_whitespace_udf(airport_df.state)\n",
    "        if airport_df.state.isNotNull\n",
    "        else airport_df.state,\n",
    "    )\n",
    "    .withColumn(\n",
    "        \"type\",\n",
    "        remove_whitespace_udf(airport_df.type)\n",
    "        if airport_df.type.isNotNull\n",
    "        else airport_df.type,\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exclude unused airport"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "used_airports = [\n",
    "    \"seaplane_base\",\n",
    "    \"medium_airport\",\n",
    "    \"small_airport\",\n",
    "    \"large_airport\",\n",
    "]\n",
    "airport_df = airport_df.filter(airport_df.type.isin(used_airports))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": []
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+------------+-----+-------------+\n",
      "|airport_id|                name|        city|state|         type|\n",
      "+----------+--------------------+------------+-----+-------------+\n",
      "|         0|        Lowell Field|Anchor Point| None|small_airport|\n",
      "|         2|Pan Lake Strip Ai...|      Willow| None|small_airport|\n",
      "|         3|     Meadow STOLport|     Jericho| None|small_airport|\n",
      "|         5|The 2A Ranch Airport|Ormond Beach| None|small_airport|\n",
      "|         7|    Pankratz Airport| Springfield| None|small_airport|\n",
      "+----------+--------------------+------------+-----+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "21/12/08 02:22:01 WARN HeartbeatReceiver: Removing executor driver with no recent heartbeats: 1177655 ms exceeds timeout 120000 ms\n",
      "21/12/08 02:22:01 WARN SparkContext: Killing executors is not supported by current scheduler.\n"
     ]
    }
   ],
   "source": [
    "airport_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 3: Define the Data Model\n",
    "#### 3.1 Conceptual Data Model\n",
    "The conceptual model has been designed keeping into the mind that the model will be used for finding effective advertisement strategies targeted for immigrants. Therefore, I have considered the immigration as a fact table, which consists of details regarding the immigrant. Further, the demographics and airports as dimension tables, since these tables will allow us to estimate the details about the movement of immigrants.\n",
    "\n",
    "Only looking at the conceptual model we could say that from this model we could easily extract information such as:\n",
    "\n",
    "- Average age of immigrant of certain visa type\n",
    "- Total number of Asian living in particular city\n",
    "- Peak month of travel\n",
    "\n",
    "The list could go on, and each insights could be effectively used for making advertisement decision.\n",
    "\n",
    "<img src=\"./images/er_diagram.png\" width=\"500\">\n",
    "\n",
    "#### 3.2 Mapping Out Data Pipelines\n",
    "\n",
    "All of the collected data were stored in the s3 data lake without transforming. In order to transform the data into the conceptual model following steps were taken\n",
    "\n",
    "- Using Amazon EMR the data stored in s3 were transformed as shown in conceptual model\n",
    "- The transformed data are than stored in s3\n",
    "- The transformed data stored in s3 is than loaded to the redshift\n",
    "- The loading of data has been monitored using Apache airflow\n",
    "\n",
    "    - Tree\n",
    "\n",
    "    <img src=\"./images/tree.png\" width=\"500\">\n",
    "\n",
    "    - Graph\n",
    "\n",
    "    <img src=\"./images/graph.png\" width=\"500\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### Step 4: Run Pipelines to Model the Data \n",
    "#### 4.1 Create the data model\n",
    "Build the data pipelines to create the data model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# In the src code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### 4.2 Data Quality Checks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Data quality checks has been performed in the airflow operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### 4.3 Data dictionary \n",
    "\n",
    "#### immigration (fact_table)\n",
    "\n",
    "|   column_name  |                              description                              | old_column_name |\n",
    "|:--------------:|:---------------------------------------------------------------------:|:---------------:|\n",
    "|      cicid     |               Unique identifier  of the immigrant record              |      cicid      |\n",
    "|   origin_city  |                      City of origin of immigrant                      |      i94cit     |\n",
    "|  traveled_from |                 City from where the immigrant traveled                |      i94res     |\n",
    "|  arrived_city  |          City where immigrant arrived, extracted from i94port         |        -        |\n",
    "|   us_address   |                       Address of immigrant in US                      |     i94addr     |\n",
    "|  arrival_date  |                      Date of arrival of immigrant                     |     arrdate     |\n",
    "| departure_date |              Date on which the immigrant departed form US             |     depdate     |\n",
    "|       age      |                          Age of the immigrant                         |       age       |\n",
    "|     gender     |                        Gender of the immigrant                        |      gender     |\n",
    "|   visa_status  |                      Visa status of the immigrant                     |     i94visa     |\n",
    "|    visa_type   |                         Visa type of immigrant                        |     visatype    |\n",
    "|   occupation   |                     Occupation of immigrant in US                     |      occup      |\n",
    "|     airline    |              Airline through which the immigrant traveled             |     airline     |\n",
    "|  matched_flag  | Flag that denotes the user has arrived and departed(or visa expired) |     mat_flag    |\n",
    "\n",
    "\n",
    "#### demographics (dim_table)\n",
    "\n",
    "| column_name                   | description                                                     |\n",
    "|-------------------------------|-----------------------------------------------------------------|\n",
    "| city_id                       | Unique id of for the city                                       |\n",
    "| city                          | Name of city                                                    |\n",
    "| state                         | State of city                                                   |\n",
    "| american_indian_alaska_native | Total population of American Indian or Alaska Native in the city |\n",
    "| asian                         | Total population of Asian in the city                           |\n",
    "| hispanic_latino               | Total population of Hispanic or Latino                          |\n",
    "| white                         | Total population of White                                       |\n",
    "| female_population             | Total number of females                                         |\n",
    "| male_population               | Total number of males                                           |\n",
    "| total_population              | Total population of the city                                    |\n",
    "\n",
    "#### airports (dim_table)\n",
    "\n",
    "| column_name | description                                                 |\n",
    "|-------------|-------------------------------------------------------------|\n",
    "| airport_id  | Unique identifier of an airport                             |\n",
    "| name        | Name of the airport                                         |\n",
    "| city        | City at which the airport is located, extracted from region |\n",
    "| state       | State at which the airport is located                       |\n",
    "| type        | Type of the airport                                         |\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "#### Step 5: Complete Project Write Up\n",
    "* Clearly state the rationale for the choice of tools and technologies for the project.\n",
    "* Propose how often the data should be updated and why.\n",
    "* Write a description of how you would approach the problem differently under the following scenarios:\n",
    " * The data was increased by 100x.\n",
    " * The data populates a dashboard that must be updated on a daily basis by 7am every day.\n",
    " * The database needed to be accessed by 100+ people."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Clearly state the rationale for the choice of tools and technologies for the project.\n",
    "\n",
    "During the development of the proposed smart advertisement system several tools and technologies were used as mentioned [above](##technology-used). I have selected each tools very carefully with consideration of project goal and time. For cloud I have selected amazon because of the following services:\n",
    "\n",
    "- Amazon s3: It is low cost, highly available, secure, infinitely scalable and easy to manage object storage service. This is perfect because effectiveness of the smart advertisement service will increases with the amount of data thus the service should increase the size of the datasets as much as possible.\n",
    "\n",
    "- Amazon EMR: This service enables us to easily transform our data using apache spark which is lightning fast analytics engine for big data.\n",
    "\n",
    "- Amazon Redshift: This service is a fast, easy, and secure cloud data warehousing service that provides high scalability. Since it is fast it is the best choice for developing api endpoints for dashboard and automatic advertisement in near future. \n",
    "\n",
    "- Managed Apache Airflow: This service allows running of airflow without maintaining servers. The airflow is extremely important tool for monitoring etl tasks. It provides a visual representation of etl process, which is highly valuable for debugging and making discussion easier with data scientist and data analyst.\n",
    "\n",
    "Furthermore, I have used notion for project management and github for the version control. These two tool are must used tools for effective delivery of a product. Also, I have used python as a programming language because I love writing codes in python.\n",
    "\n",
    "#### Propose how often the data should be updated and why ?\n",
    "\n",
    "The data should be updated once a month because it would be difficult get data from US National Tourism and Trade Office often since the data has to be requested. Also, in the real product we will be using huge amount, performing analysis on one month old data might have nearly same effectiveness.\n",
    "\n",
    "#### How to approach the problem if the data was increased by 100x.\n",
    "\n",
    "- Transforming data: Currently in EMR only one master and slave node are running, since the EMR is highly scalable, I would have used appropriate number of slaves. I would also have partitioned the immigration data by month while writing in parquet format.\n",
    "\n",
    "- Loading data to redshift: I would have increase the size of the warehouse and while copying the data from s3 to redshift I would have used template filed that would allow to load timestamped files from S3 based on the execution time and run backfills.\n",
    "\n",
    "#### The data populates a dashboard that must be updated on a daily basis by 7am every day.\n",
    "\n",
    "- Use airflow to schedule and run pipelines\n",
    "\n",
    "#### The database needed to be accessed by 100+ people.\n",
    "\n",
    "- I would have used developed endpoints using sql client, lambda function and api gateway.\n"
   ]
  }
 ],
 "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.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}