<jupyter_start><jupyter_text>Exploring the Influence of City and Temperature on US Immigration Data Engineering Capstone Project Project SummaryThis capstone project creates an analytics database to support end user queries exploring the influences of destination city and temperature on US immigration. Potential user queries include:* Visitors by age, gender, visa category, and country of origin.* Relationship between U.S state destination and country of origin.Future user queries might include:* Relationship between U.S state destination and climate in country of origin.* Relationship between U.S. state demographics and country of origin.* Relationship between U.S. state demographics and climate in country of origin.The project follows the following steps:* Step 1: Scope the Project and Gather Data* Step 2: Explore and Assess the Data* Step 3: Define the Data Model* Step 4: Run ETL to Model the Data* Step 5: Complete Project Write Up<jupyter_code># Import required libraries and modules
import pandas as pd
from pyspark.sql import SparkSession<jupyter_output><empty_output><jupyter_text>Step 1: Scope the Project and Gather Data Scope The project involves building an ETL pipeline that extracts the data from the four datasets, processes them using Spark, and loads the data into as a set of dimensional tables based on the star schema. The data originates from four datasets. The main dataset includes data on U.S. immigration, while supplementary datasets include data on airport codes, U.S. city demographics, and temperature data. Data Description U.S. Immigration Data:- **Data Source**: The U.S. immigration data originates from the [National Travel & Tourism Office](https://travel.trade.gov/research/reports/i94/historical/2016.html). - **Description**: Each report contains international visitor arrival statistics by world regions and select countries (including top 20), type of visa, mode of transportation, age groups, states visited (first intended address only), and the top ports of entry (for select countries). Data sources include: Overseas DHS/CBP I-94 Program data; Canadian visitation data (Stats Canada) and Mexican visitation data (Banco de Mexico). - **Information Included**: Udacity has provided 2016 data in the folder ```../../data/18-83510-I94-Data-2016/``` in SAS7BDAT binary storage format. Udacity has also provided April 2016 data in the folder ```./sas_data``` and sample data in the file ```./immigration_data_sample.csv```. The April 2016 files have 28 columns and 3,096,313 rows of data in Snappy Parquet format, while the sample data file has 28 columns and 1,000 rows of data in CSV format. Data is also available for purchase from the National Travel & Tourism Office website.U.S. City Demographic Data: - **Data Source**: The U.S. city demographic data originates from [OpenSoft](https://public.opendatasoft.com/explore/dataset/us-cities-demographics/export/).- **Description**: This dataset contains information about the demographics of all US cities and census-designated places with a population greater or equal to 65,000. This data comes from the US Census Bureau's 2015 American Community Survey.- **Information Included**: Udacity has provided sample data in the following path: ```./us-cities-demographics.csv```. The file has 12 columns and 2,891 rows of data. Data is also available from OpenSoft in CSV, JSON, or Excel formats. Global Temperature Data:- **Data Source**: The global temperature data originates from Kaggle's [Climate Change: Earth Surface Temperature Data](https://www.kaggle.com/berkeleyearth/climate-change-earth-surface-temperature-data) dataset.- **Description**: Kaggle has repackaged the data from a newer compilation put together by the Berkeley Earth, which is affiliated with Lawrence Berkeley National Laboratory. The Berkeley Earth Surface Temperature Study combines 1.6 billion temperature reports from 16 pre-existing archives. The data starts in 1750 for average land temperature and 1850 for max and min land temperatures and global ocean and land temperatures.- **Information Included**: Udacity has provided data in a folder with the following path: ```../../data2/```. There's just one file in that folder, called ```GlobalLandTemperaturesByCity.csv```. The file has 7 columns and 8,599,212 rows of data in CSV format. Data is also available from the Kaggle dataset, which inlcudes 4 additional files:  - GlobalLandTemperaturesByCountry.csv  - GlobalLandTemperaturesByMajorCity.csv  - GlobalLandTemperaturesByState.csv  - GlobalLandTemperaturesByState.csvWorld Airport Code Table: - **Data Source**: The world airport code table originates from [DataHub](https://datahub.io/core/airport-codesdata).- **Description**: The airport codes may refer to either IATA airport code, a three-letter code which is used in passenger reservation, ticketing and baggage-handling systems, or the ICAO airport code which is a four letter code used by ATC systems and for airports that do not have an IATA airport code (from wikipedia). Airport codes from around the world. Downloaded from public domain source http://ourairports.com/data/ who compiled this data from multiple different sources. This data is updated nightly.- **Information Included**: Udacity has provided the data in the following path: ```./airport-codes_csv.csv```. The file has 12 columns and 55,075 rows of data in CSV format. Data is also available from DataHub.  Data Gathering<jupyter_code># Read in the data here, including sample immigration data
df_immigration_sample = pd.read_csv('./immigration_data_sample.csv')
df_demographics = pd.read_csv('./us-cities-demographics.csv', sep=';')
df_temperatures = pd.read_csv('../../data2/GlobalLandTemperaturesByCity.csv')
df_airport_codes = pd.read_csv('./airport-codes_csv.csv')
# Display first 5 rows of each dataset
print(df_immigration_sample.shape)
df_immigration_sample.head()
print(df_demographics.shape)
df_demographics.head()
print(df_temperatures.shape)
df_temperatures.head()
print(df_airport_codes.shape)
df_airport_codes.head()
# Create Spark session to read in April 2016 immigration dataset
from pyspark.sql import SparkSession
spark = SparkSession.builder.\
config("spark.jars.packages","saurfang:spark-sas7bdat:2.0.0-s_2.11")\
.enableHiveSupport().getOrCreate()
df_immigration=spark.read.parquet("sas_data")
print(df_immigration.count())
df_immigration.head(5)<jupyter_output>3096313<jupyter_text>Step 2: Explore and Assess the Data Explore the Data Identify data quality issues, like missing values, duplicate data, etc.  **U.S. Immigration Data:**<jupyter_code># Immigration data
df_immigration.printSchema()
df_immigration.describe().show()<jupyter_output>root
 |-- cicid: double (nullable = true)
 |-- i94yr: double (nullable = true)
 |-- i94mon: double (nullable = true)
 |-- i94cit: double (nullable = true)
 |-- i94res: double (nullable = true)
 |-- i94port: string (nullable = true)
 |-- arrdate: double (nullable = true)
 |-- i94mode: double (nullable = true)
 |-- i94addr: string (nullable = true)
 |-- depdate: double (nullable = true)
 |-- i94bir: double (nullable = true)
 |-- i94visa: double (nullable = true)
 |-- count: double (nullable = true)
 |-- dtadfile: string (nullable = true)
 |-- visapost: string (nullable = true)
 |-- occup: string (nullable = true)
 |-- entdepa: string (nullable = true)
 |-- entdepd: string (nullable = true)
 |-- entdepu: string (nullable = true)
 |-- matflag: string (nullable = true)
 |-- biryear: double (nullable = true)
 |-- dtaddto: string (nullable = true)
 |-- gender: string (nullable = true)
 |-- insnum: string (nullable = true)
 |-- airline: string (nullable = true)
 |-- admnum: double (nullable = [...]<jupyter_text>- Data Dictionary (from I94_SAS_Labels_Descriptions.SAS):| Column Name | Description ||-----|-----|| cicid | Unique identifier || i94yr | 4 digit year of entry || i94mon | Numeric month of entry || i94cit | Country of citizenship (list of valid and invalid 3-digit country codes) || i94res | Country of residence (list of valid and invalid 3-digit country codes) || i94port | U.S. port of entry (list of valid and invalid 3-letter city/town codes) || arrdate | Arrival Date in the USA. It is a SAS date numeric field. || i94mode | Mode of transport: 1 = 'Air', 2 = 'Sea', 3 = 'Land', 9 = 'Not reported' || i94addr | Arrival address (list of valid and invalid 2-letter state codes) || depdate | Departure Date from the USA. It is a SAS date numeric field. || i94bir | Age of Respondent in Years || i94visa | Visa codes collapsed into three categories: 1 = Business, 2 = Pleasure, 3 = Student || count | Used for summary statistics || dtadfile |  Character Date Field - Date added to I-94 Files - CIC does not use || visapost | Department of State where where Visa was issued - CIC does not use || occup | Occupation that will be performed in U.S. - CIC does not use || entdepa | Arrival Flag - admitted or paroled into the U.S. - CIC does not use || entdepd | Departure Flag - Departed, lost I-94 or is deceased - CIC does not use | | entdepu | Update Flag - Either apprehended, overstayed, adjusted to perm residence - CIC does not use | | matflag | Match flag - Match of arrival and departure records | | biryear | 4 digit year of birth | | dtaddto | Character Date Field - Date to which admitted to U.S. (allowed to stay until) - CIC does not use | | gender | Non-immigrant sex  | | insnum | INS number | | airline | Airline used to arrive in U.S. | | admnum | Admission Number || fltno | Flight number of Airline used to arrive in U.S. || visatype | Class of admission legally admitting the non-immigrant to temporarily stay in U.S. | - Observations:  - Fields with a high number of missing values include: visapost, occup, entdepu, and insnum. These fields should be discarded.  - Fields not providing useful information for our purposes include: count, dtadvile, visapost, entdepa, entdepd, entdepu, matflag, dtaddto, insnum, and fltno. These fields should be discarded.  - Ages (i94bir) range from -3 to 114. These records should be filtered to exclude ages 100, which can be assumed to be input errors.  - The dates are stored in SAS date format (days since January 1, 1960). These dates should be converted YYYY-MM-DD string format.  **U.S. City Demographics Data:**<jupyter_code># City demographics data
df_demographics.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 2891 entries, 0 to 2890
Data columns (total 12 columns):
City                      2891 non-null object
State                     2891 non-null object
Median Age                2891 non-null float64
Male Population           2888 non-null float64
Female Population         2888 non-null float64
Total Population          2891 non-null int64
Number of Veterans        2878 non-null float64
Foreign-born              2878 non-null float64
Average Household Size    2875 non-null float64
State Code                2891 non-null object
Race                      2891 non-null object
Count                     2891 non-null int64
dtypes: float64(6), int64(2), object(4)
memory usage: 271.1+ KB<jupyter_text>- Data Dictionary:| Column Name | Description ||-----|-----|| City | Name of city || State | Name of state || Median Age | Median age of city population || Male Population | Number of males living in city || Female Population | Number of females living in city || Total Population | Total number of people living in city || Number of Veterans | Number of veterans living in city || Foreign-born | Number of non-U.S. born people living in city || Average Household Size | Average number of people per household in city || State Code | Code of state || Race | Race class of people living in city || Count | Number of people in race class living in city | - Observations:  - There are no fields that contain a significant number of missing values.  - For this particular project, the decision has been made to not use this dataset.  **World Temperature Data:**<jupyter_code># Temperature data
df_temperatures.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 8599212 entries, 0 to 8599211
Data columns (total 7 columns):
dt                               object
AverageTemperature               float64
AverageTemperatureUncertainty    float64
City                             object
Country                          object
Latitude                         object
Longitude                        object
dtypes: float64(2), object(5)
memory usage: 459.2+ MB<jupyter_text>- Data Dictionary:| Column Name | Description ||-----|-----|| dt | Date of temperature reading (YYYY-MM-DD) || AverageTemperature | Average temperature in city on date || AverageTemperatureUncertainty | Uncertainty measure of average temperature || City | Name of city || Country | Number of country || Latitude | Latitude of city || Longitude | Longitude of city | - Observations:  - Fields not providing useful information for our purposes include: AverageTemperatureUncertainty, Latitude, and Longitude. These fields should be discarded.  - For this particular project, the decision has been made to not use this dataset.  **Global Airport Codes Data:**<jupyter_code># Airport codes data
df_airport_codes.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 55075 entries, 0 to 55074
Data columns (total 12 columns):
ident           55075 non-null object
type            55075 non-null object
name            55075 non-null object
elevation_ft    48069 non-null float64
continent       27356 non-null object
iso_country     54828 non-null object
iso_region      55075 non-null object
municipality    49399 non-null object
gps_code        41030 non-null object
iata_code       9189 non-null object
local_code      28686 non-null object
coordinates     55075 non-null object
dtypes: float64(1), object(11)
memory usage: 5.0+ MB<jupyter_text>- Data Dictionary:| Column Name | Description ||-----|-----|| ident | Unique identifier || type | Type of airport || name | Name of airport || elevation_ft | Altitude of airport (feet) || continent | Continent of airport || iso_country | ISO country code of airport (2 letters) || iso_region | ISO region code of airport (4 letters) || municipality | City of airport || gps_code | GPS code for airport || iata_code | IATA code for airport (3 letters) || local_code | ICAO code for airport (4 letters)|| coordinates | GPS coordinates for airport | - Observations:  - Fields with a high number of missing values include: continent, iata_code, and local_code.   - Fields not providing useful information for our purposes include: type, elevation_ft, continent, iso_country, iso_region, municality, gps_code, and coordinates.  Cleaning Steps- U.S. Immigration Data:  - Drop records if duplicates exist.  - Drop records with NaN values in key columns.- U.S. City Demographics Data:  - Drop duplicate records.  - Drop records with NaN values.- World Temperature Data:  - Drop duplicate records.  - Drop records with NaN values in key columns.- Global Airport Codes Data:  - Drop duplicate records.  - Drop records with NaN values in key columns.<jupyter_code># Clean U.S. immigration data

def clean_immig(df_immig):
    '''
    Performs cleaning tasks on U.S. immigration dataset
    Input: original dataframe
    Output: transformed dataframe
    '''
    # Drop records if duplicates exist
    if df_immig.count() != df_immig.dropDuplicates().count():
        df_immig.dropDuplicates()
        
    # Drop records with NaN values in key columns
    df_immig = df_immig.dropna(subset=['i94yr', 'i94mon', 'i94res','i94port', 'arrdate', 'i94bir', 'gender', 'admnum'])
    
    return df_immig
# Clean U.S. city demographics data

def clean_demo(df_demo):
    '''
    Performs cleaning tasks on U.S. city demographics dataset
    Input: original dataframe
    Output: transformed dataframe
    '''    
    # Drop duplicate records
    df_demo.drop_duplicates()
    
    # Drop records with NaN values
    df_demo.dropna()
    
    return df_demo
# Clean world temperature data

def clean_temp(df_temp):
    '''
    Performs cleaning tasks on world temperature dataset
    Input: original dataframe
    Output: transformed dataframe
    '''
    # Drop duplicate records
    df_temp.drop_duplicates()
    
    # Drop records with NaN values in key columns
    df_temp = df_temp.dropna(subset=['dt', 'AverageTemperature', 'City', 'Country'])
    
    return df_temp
# Clean global airport data

def clean_airport(df_airport):
    '''
    Performs cleaning tasks on global airport codes dataset
    Input: original dataframe
    Output: transformed dataframe
    '''
    # Drop duplicate records
    df_airport.dropDuplicates()
    
    # Drop records with NaN values in key columns
    df_airport = df_airport.dropna(subset=['ident', 'name', 'iso_region'])
    
    return df_airport<jupyter_output><empty_output><jupyter_text>Step 3: Define the Data Model 3.1 Conceptual Data ModelThe data model is built around a specific business process, the decision by U.S. Immigration to admit visitors into the United States of America. This process generates immigration events and related performance metrics.The STAR schema design, one of the most widely-used schemas in the industry, is also the simplest style of data mart schema. It was selected for this situation because, compared to other potential choices (e.g. SNOWFLAKE Schema):1. It is more effective at handling simplified queries. As this data mart has just been created, it is likely that U.S. Immigration BI user needs will be basic. 2. It supports fast aggregations. When simple queries are used, performance can be improved, which is important given a potential large dataset of historical immigration activity.3. It supports simplified reporting. Business reporting logic, such as period-over-period and as-of reporting, is generally much simpler for the star schema.Dimension tables should provide information on the *who, what, where, when, why, and how* related to the U.S. immigration process. Therefore, the following dimension tables have been built:- **dim_immigrant** provides information on *who* is entering the U.S.- **dim_aiport** provides information on *how* this person entered the U.S.- **dim_country** provides information on *where* this person originates.- **dim_time** provides information on *when* this person entered the U.S.The fact table should contain measurements and metrics related to the U.S. immigration process. Here, the following fact table has been built:- **fact_immigration** provides the measurements and metrics related to the U.S. immigration process, e.g. how many people entered the U.S.?Following is an Entity Relationship Diagram (ERD) depicting the STAR schema used for this project:![data model](data_model.png)  3.2 Mapping Out Data PipelinesList of steps necessary to pipeline the data into the chosen data model:1. Extract U.S. immigration data to create a Spark dataframe.2. Extract Global airport code data to create Pandas dataframe.3. Clean immigration and airport data by removing duplicates and NaN records.4. Create dimension tables (airport, country, immigrant, time) and write to parquet files.5. Create a fact table (immigration) and write to parquet file.  Step 4: Run Pipelines to Model the Data  4.1 Create the data modelBuild the data pipelines to create the data model.<jupyter_code># Import required modules and libraries
import os
import glob
import json
import pandas as pd
import pyspark.sql.functions as f
from datetime import datetime, timedelta, date
from pyspark.sql.types import StringType
from pyspark.sql import SparkSession
from pyspark.sql import types as t
from pyspark.sql.functions import udf
from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear<jupyter_output><empty_output><jupyter_text>4.1.1 Extract and Clean Immigration Data<jupyter_code># Create Spark session
spark = SparkSession \
        .builder \
        .config('spark.jars.packages','saurfang:spark-sas7bdat:2.0.0-s_2.11') \
        .enableHiveSupport().getOrCreate()
# Load immmigration labels
#dir_path = os.path.dirname(os.path.realpath(__file__))
#immig_labels = json.load(open(dir_path + '/immigration_labels.json'))
immig_labels = json.load(open('./immigration_labels.json'))

country_code = immig_labels['country_code']
city_code = immig_labels['city_code']
travel_code = immig_labels['travel_code']
state_list = immig_labels['state_code']
state_code = {state: code for code, state in state_list.items()}
visa_code = immig_labels['visa_code']
# Create user defined functions for codes
country_udf = udf(lambda country: country_code[country], StringType())
city_udf = udf(lambda city: city_code[city], StringType())
travel_udf = udf(lambda mode: travel_code[mode], StringType())
state_udf = udf(lambda state: state_code[state], StringType())
visa_udf = udf(lambda visa: visa_code[visa], StringType())
# Get filepath to immigration source and staging data files
immigration_source = '../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat'
immigration_staging = 'staging_immigration'
# Read immigration data file
df_immigration = spark.read.format('com.github.saurfang.sas.spark').load(immigration_source)
# Clean immigration data file
df_immigration_clean = clean_immig(df_immigration)
# Define timestamp column
@udf(t.TimestampType())
def get_timestamp (arrdate):
    arrdate_int = int(arrdate)
    return (datetime(1960,1,1) + timedelta(days=arrdate_int))
# Create staging file against which SQL queries can be run
df_immigration_stage = df_immigration_clean.filter(df_immigration_clean.i94addr.isNotNull())\
    .filter(df_immigration_clean.i94res.isNotNull())\
    .filter(f.col("i94addr").isin(list(state_code.keys())))\
    .filter(f.col("i94port").isin(list(city_code.keys())))\
    .withColumn('immigration_id', f.col('cicid').cast('integer'))\
    .withColumn('country_code', f.col('i94res').cast('integer').cast('string'))\
    .withColumn('arrival_date', f.col('arrdate'))\
    .withColumn('arrival_time', get_timestamp('arrival_date'))\
    .withColumn('country', country_udf(f.col('country_code')))\
    .withColumn('state_code', f.col('i94addr'))\
    .withColumn('state', state_udf(f.col('i94addr')))\
    .withColumn('airport_id', f.col('i94port'))\
    .withColumn('city', city_udf(f.col('i94port')))\
    .withColumn('arrival_year', f.col('i94yr').cast('integer'))\
    .withColumn('arrival_month', f.col('i94mon').cast('integer'))\
    .withColumn('arrival_mode_code', f.col('i94mode').cast('integer').cast('string'))\
    .withColumn('arrival_mode', travel_udf(f.col('arrival_mode_code')))\
    .withColumn('age', f.col('i94bir').cast('integer'))\
    .withColumn('visa_category_code', f.col('i94visa').cast('integer').cast('string'))\
    .withColumn('visa_category', visa_udf(f.col('visa_category_code')))\
    .withColumn('gender', f.col('gender'))\
    .withColumn('admission_id', f.col('admnum'))
# Write staging file to parquet files
df_immigration_stage.select('immigration_id', 'arrival_date', 'arrival_time', 'country_code', 'country', 'state_code', 
                            'state', 'airport_id', 'city', 'arrival_year', 'arrival_month', 'arrival_mode_code', 
                            'arrival_mode', 'age', 'visa_category_code', 'visa_category', 'gender', 'admission_id')\
                    .write.mode('overwrite')\
                    .option('compression', 'gzip')\
                    .parquet(immigration_staging)<jupyter_output><empty_output><jupyter_text>4.1.2 Extract and Clean Airport Data<jupyter_code># Get filepath to airport source and staging data files
airport_source = './airport-codes_csv.csv'
airport_staging = 'staging_airport'
# Read airport data file
df_airport_codes = spark.read.csv(airport_source, header=True)
# Clean airport data file
df_airport_codes_clean = clean_airport(df_airport_codes)
# Replace iso_region with state code
split_col = f.split(df_airport_codes_clean['iso_region'], '-')
# Create staging file against which SQL queries can be run
df_airport_stage = df_airport_codes_clean\
    .withColumn('airport_id', f.col('ident'))\
    .withColumn('airport_type', f.col('type'))\
    .withColumn('airport', f.col('name'))\
    .withColumn('state_code', split_col.getItem(1))\
    .withColumn('city', f.col('municipality'))
# Write staging file to parquet files
df_airport_stage.select('airport_id', 'airport_type', 'airport', 'state_code', 'city')\
                    .write.mode('overwrite')\
                    .option('compression', 'gzip')\
                    .parquet(airport_staging)<jupyter_output><empty_output><jupyter_text>4.1.3 Create dimension tables<jupyter_code># Read immigration staging parquet files
df_immigration = spark.read.parquet(immigration_staging)
# Create a temporary view against which SQL queries can be run
df_immigration.createOrReplaceTempView('immigrant')
# Create immigrant dimension table
dim_immigrant = spark.sql("""
    SELECT DISTINCT
        immigrant.admission_id,
        immigrant.country_code,
        immigrant.age,
        immigrant.gender,
        immigrant.visa_category,
        immigrant.visa_category_code
    FROM immigrant
    ORDER BY 1,2
""").cache()
# Inspect immigrant dimension table schema and first 20 rows
dim_immigrant.printSchema()
dim_immigrant.show()
# Write immigrant table to parquet file
dim_immigrant.write.mode("overwrite").parquet('dim_immigrant')
# Create a temporary view against which SQL queries can be run
df_immigration.createOrReplaceTempView('country')
# Create country dimension table
dim_country = spark.sql("""
    SELECT DISTINCT
        country.country_code,
        country.country
    FROM country
    ORDER BY 1
""").cache()
# Inspect country dimension table schema and first 20 rows
dim_country.printSchema()
dim_country.show()
# Write country table to parquet file
dim_country.write.mode("overwrite").parquet('dim_country')
# Create a temporary view against which SQL queries can be run
df_immigration.createOrReplaceTempView('time')
# Create time dimension table
dim_time = spark.sql("""
    SELECT DISTINCT
        time.arrival_time  AS arrival_ts, 
        hour(arrival_time)       AS hour, 
        day(arrival_time)        AS day, 
        weekofyear(arrival_time) AS week,
        month(arrival_time)      AS month,
        year(arrival_time)       AS year,
        dayofweek(arrival_time)  AS weekday
    FROM time
""").cache()
# Inspect time dimension table schema and first 20 rows
dim_time.printSchema()
dim_time.show()
# Write time table to parquet file
dim_time.write.mode("overwrite").parquet('dim_time')
# Read airport staging parquet files
df_airport = spark.read.parquet(airport_staging)
# Create a temporary view against which SQL queries can be run
df_airport.createOrReplaceTempView('airport')
# Create airport dimension table
dim_airport = spark.sql("""
    SELECT DISTINCT
        airport.airport_id, 
        airport.airport_type, 
        airport.airport, 
        airport.state_code, 
        airport.city 
    FROM airport
""").cache()
# Inspect airport dimension table schema and first 20 rows
dim_airport.printSchema()
dim_airport.show()
# Write airport table to parquet file
dim_airport.write.mode("overwrite").parquet('dim_airport')<jupyter_output><empty_output><jupyter_text>4.1.4 Create fact table<jupyter_code># Create a temporary view against which SQL queries can be run
df_immigration.createOrReplaceTempView('immigration')
# Inspect fact table schema and first 20 rows
df_immigration.printSchema()
# Create fact table
fact_immigration = spark.sql("""
    SELECT DISTINCT
        immigration.immigration_id,
        immigration.arrival_time,
        immigration.admission_id,
        immigration.airport_id,
        immigration.country_code,
        immigration.arrival_mode,
        immigration.arrival_year,
        immigration.arrival_month
    FROM immigration
""").cache()
# Inspect fact table schema and first 20 rows
fact_immigration.printSchema()
fact_immigration.show()
# Write immigration table to parquet file
fact_immigration.write.mode("overwrite").parquet('fact_immigration')<jupyter_output><empty_output><jupyter_text>4.2 Data Quality ChecksExplain the data quality checks you'll perform to ensure the pipeline ran as expected. These could include: * Integrity constraints on the relational database (e.g., unique key, data type, etc.) * Unit tests for the scripts to ensure they are doing the right thing * Source/Count checks to ensure completeness Run Quality Checks<jupyter_code># Load tables to be checked
fact_immigration = spark.read.parquet('fact_immigration')
dim_immigrant = spark.read.parquet('dim_immigrant')
dim_country = spark.read.parquet('dim_country')
dim_time = spark.read.parquet('dim_time')
dim_airport = spark.read.parquet('dim_airport')
# Define zero record check
def zero_check(df, description):
    '''
    Input: Spark fact or dimension tables
    Output: Outcome of quality check
    '''
    result = df.count()
    if result == 0:
        print("Data quality check failed for {} with zero records".format(description))
    else:
        print("Data quality check passed for {} with {} records".format(description, result))
    return
# Perform zero record check
zero_check(fact_immigration, 'immigration fact table')
zero_check(dim_immigrant, 'immigrant dimension table')
zero_check(dim_country, 'country dimension table')
zero_check(dim_time, 'time dimension table')
zero_check(dim_airport, 'airport dimension table')
# Perform null value check
null_values_check = fact_immigration.select(f.isnull('immigration_id'),\
                  f.isnull('arrival_time'),\
                  f.isnull('admission_id'),\
                  f.isnull('airport_id'),\
                  f.isnull('country_code'),\
                  f.isnull('arrival_mode'),\
                  f.isnull('arrival_year'),\
                  f.isnull('arrival_month'))

null_values_check.show()<jupyter_output>+------------------------+----------------------+----------------------+--------------------+----------------------+----------------------+----------------------+-----------------------+
|(immigration_id IS NULL)|(arrival_time IS NULL)|(admission_id IS NULL)|(airport_id IS NULL)|(country_code IS NULL)|(arrival_mode IS NULL)|(arrival_year IS NULL)|(arrival_month IS NULL)|
+------------------------+----------------------+----------------------+--------------------+----------------------+----------------------+----------------------+-----------------------+
|                   false|                 false|                 false|               false|                 false|                 false|                 false|                  false|
|                   false|                 false|                 false|               false|                 false|                 false|                 false|                  false|
|                   false|                 false|                [...]
{'fraction_non_alphanumeric': 0.1060159865376525, 'fraction_numerical': 0.016407236011779555, 'mean_word_length': 3.8014294592914855, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 50, 'https://': 4, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '333076', 'license': 'MIT', 'n_tokens_mistral': 9020, 'n_tokens_neox': 8408, 'n_words': 3118, 'path': 'course-5-capstone-project/project-data-engineering-capstone/Capstone Project.ipynb', 'repo_name': 'robstraker/data-engineering-nanodegree-udacity', 'size': '30901'}