<jupyter_start><jupyter_text>Data Engineering Capstone Project Project SummaryThe goal of this project is to create an ETL pipeline using I94 immigration and city temperature data to create a database optimized for queries on immigration events. This database can be used to answer questions that are related to immigration behavior.The project follows these 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 UpPlease note that this notebook is designed to be run on Udacity's notebook workspace as the data is provided internally by Udacity.  Also note that we will be using Pandas Redshift as a convenience method to upload our Pandas dataframes as tables to Amazon Redshift.<jupyter_code>!pip install pandas_redshift
# Do all imports and installs here
import pandas as pd
import pandas_redshift as pr
import datetime
import re<jupyter_output><empty_output><jupyter_text>Step 1: Scope the Project and Gather Data Scope The intent of this project is to build a data warehouse on AWS Redshift and I plan on designing an ETL pipeline to load data into tables for OLAP.  The data that will be folded in will include sources from the I94 US Immigration dataaset provided by Udacity, as well as data that involves airport codes, US city demographics and data regarding temperature. Describe and Gather Data The I94 immigration [data](https://travel.trade.gov/research/reports/i94/historical/2016.html) origtinates from the US National Tourism and Trade Office.  An additional intricacy is that is provided in SAS7BDAT [format](https://cran.r-project.org/web/packages/sas7bdat/vignettes/sas7bdat.pdf) and is a binary database storage format.  It 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).  There are 12 parts of this datasets with over 40 million records combined.  However we will concentrate on the data that comes from April 2016 that consists of roughly 3 million records.  Some relevant attributes include:* `i94yr` - 4 digit year* `i94mon` - Numeric month* `i94cit` - 3 digit code of origin city* `i94port` - 3 character code of destination USA city* `arrdate` - Arrival date in the USA* `i94mode` - 1 digit travel code* `depdate` - Departure date from the USA* `i94visa` - Reason for immigration<jupyter_code># Read in the data here - This takes some time to run.  Be patient.
fname = '../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat'
df = pd.read_sas(fname, 'sas7bdat', encoding="ISO-8859-1")
# Show the first 5 elements of this dataframe
df.head()<jupyter_output><empty_output><jupyter_text>Relevant Column Names to be used in our analysisColumn Name|Description---|---*cicid*|ID that uniquely identify one record in the dataseti94yr | 4 digit yeari94mon | Numeric monthi94cit | 3 digit code of source city for immigration (Birth country)i94res | 3 digit code of source country for immigration (Residence country)i94port | Port admitted througharrdate | Arrival date in the USAi94mode | Mode of transportation (1 = Air, 2 = Sea, 3 = Land, 9 = Not reported)i94addr | State of arrivaldepdate | Departure datei94bir | Age (years) of respondenti94visa | Visa codes broken down into 3 categories: (1 = Business; 2 = Pleasure; 3 = Student)count | Used for summary statisticsdtadfile | Character Date Fieldvisapost | Department of State where where Visa was issuedoccup | Occupation to be performed in U.S.entdepa | Arrival Flag - Admitted or parolled into the USentdepd | Departure Flag - Departed, lost visa, or deceasedentdepu | Update Flag - Update of visa, either apprehended, overstayed, or updated to permanent residentmatflag | Match flagbiryear | 4 digit year of birthdtaddto | Character date field to when admitted to the USgender | Genderinsnum | INS numberairline | Airline used to arrive in U.S.admnum | Admission number - Should be unique and not nullfltno | Flight number of the airline used to arrive in the USvisatype | Class of admission legally allowing the non-immigrant to temporarily stay in US  US City Demographic Data[This dataset](https://public.opendatasoft.com/explore/dataset/us-cities-demographics/export/) contains various information about the demographics of all US cities and census-designated places with a population greater or equal to 65,000.  This comes from the 2015 American Community Survey from the US Census Bureau's 2015 American Community Survey.<jupyter_code># Read in the US city demographics data and show the first 5 rows of the dataframe
us_city_demographics_df = pd.read_csv('us-cities-demographics.csv', delimiter=';')
us_city_demographics_df.head()<jupyter_output><empty_output><jupyter_text>Step 2: Explore and Assess the Data Explore the Data  *I94 Immigration Data*The following columns have missing data:* `i94mode`* `i94bir`* `dtadfile`* `visapost`* `occup`* `entdepa`* `entdepd`* `entdepu`* `matflag`* `biryear`* `dtaddto`* `gender`* `insnum`* `airline`* `fltno`Due to this being missing, we will drop these columns.  The most important columns we will keep are:* `cicid` (naturally)* `i94yr`* `i94mon`,* `i94cit`* `i94res`* `i94port`* `arrdate`* `i94addr`* `depdate`* `i94visa`* `count`* `admnum`* `visatype`Let's display the total number of records for each column below.  Take note that we are dealing with 3,096,313 total records.  Any columns that don't have this amount are deemed missing and are removed from further analysis.<jupyter_code># Show non-null counts for each column of the immigration data
df.count()<jupyter_output><empty_output><jupyter_text>*US City Demographic Data*The following columns have missing data:* `Male Population`* `Female Population`* `Number of Veterans`* `Foreign-born`* `Average Household Size` We only interested in the `Total Population` in any case.  Take note that there are 2891 records for this dataset so any columns that don't match this should be removed.<jupyter_code># Show the non-null counts for each column in the demographics data
us_city_demographics_df.count()<jupyter_output><empty_output><jupyter_text>Cleaning Steps  I94 Immigration DataAs noted above, we will select out the relevant columns and remove the columns that are missing.<jupyter_code># Select out the relevant columns and drop any rows that are missing
us_immigration_df = df[['cicid', 'i94yr', 'i94mon', 'i94cit', 'i94res', 'i94port', 'arrdate', 'i94addr', 'depdate', 'i94visa', 'count', 'admnum', 'visatype']]
us_immigration_df = us_immigration_df.dropna()

# There is an escape character with the i94addr column
# Remove this using regular expressions where any repeated whitespace is removed
us_immigration_df['i94addr'] = us_immigration_df['i94addr'].map(lambda col: re.sub('\W+', '', col))
# Show the first 5 elements of this new dataframe
us_immigration_df.head()<jupyter_output><empty_output><jupyter_text>US City Demographic DataAs noted above, we select out the relevant columns.  Also note that we should drop duplicates!<jupyter_code># Select out the relevant columns and drop any duplicates
us_city_demographics_df = us_city_demographics_df[['State Code', 'City', 'State', 'Total Population']].drop_duplicates()

# Show the first 5 elements of this new dataframe
us_city_demographics_df.head()<jupyter_output><empty_output><jupyter_text>Step 3: Define the Data Model 3.1 Conceptual Data ModelAs we naturally learned in this course, we will use a more denormalized version of this combined data to minimize the amount of joins and to allow our data model to be used and interpreted in a more meaningful way.  Specifically, I chose to use the star schema with the immigration data serving as the fact table while the data that deals with dates and the states themselves are the dimension tables.  This will ultimately allow us to aggregate information about immigration based on fine grained time units, such as month or year as well the state itself. Fact Table`immigration`- Columns: *`cicid`*, `i94yr`, `i94mon`, `i94cit`, `i94res`, `i94port`, `arrdate`, `i94addr`, `depdate`, `i94visa`, `count`, `admnum`, `visatype``cicid` will serve as the primary key Dimension Tables`date` - arrival and departure date in immigration broken down into specific units - *`sas_date`*, date, day, month, year, weekday`sas_date` will serve as the primary key.  This is a custom column that will be created here.  The SAS date is a value that represents the number of days between January 1, 1960, and a specified date.`state` - total population of the states- *`state_code`*, state, total_population`state_code` will serve as the primary key.  3.2 Mapping Out Data PipelinesList the steps necessary to pipeline the data into the chosen data modelThis will be outlined below.  Step 4: Run Pipelines to Model the Data  4.1 Create the data modelBuild the data pipelines to create the data model.  Create the state table<jupyter_code># We group all city population by state and aggregate by summing over them
state_pop = us_city_demographics_df.groupby('State').sum().reset_index()
state_pop.head()
# Perform a right join to merge the state code and state together so we establish a mappin
# between state code, the full name of the state as well as its population.
state_pop_df = pd.merge(us_city_demographics_df[['State Code', 'State']].drop_duplicates(), state_pop, how='right', on='State')
state_pop_df.columns = ['state_code', 'state', 'population']

# Note we will rename these columns so that we can write to Redshift later for easy access
state_pop_df.head()<jupyter_output><empty_output><jupyter_text>Create the date tableFor the SAS date, we will convert this into a readable date format that will allow us to do datetime processing with ease.  We'll do the same foro the departure date as well.We extract all values from arrdate and depdate and convert them into readable date format.We will first get the arrival date as a table in readable form and decompose the date into further units such as weekday, day, month and year.<jupyter_code># Make a copy of the arrdate column for us to manipulate
arrdate = us_immigration_df[['arrdate']].copy()
# Rename the arrdate column to sas_date for final storage on Redshift
arrdate = arrdate.rename(columns={"arrdate": "sas_date"})

# Make this new column so that it provides date in readable format
arrdate['date'] = arrdate['sas_date'].apply(lambda col: datetime.date(1960, 1, 1) + datetime.timedelta(days=col))

# Also put in the weekday, day, month and year as new columns
arrdate['weekday'] = arrdate['date'].apply(lambda col: col.weekday())
arrdate['day'] = arrdate['date'].apply(lambda col: col.day)
arrdate['month'] = arrdate['date'].apply(lambda col: col.month)
arrdate['year'] = arrdate['date'].apply(lambda col: col.year)

# Drop duplicates
arrdate = arrdate.drop_duplicates()
print(arrdate.shape)
arrdate.head()<jupyter_output>(30, 6)<jupyter_text>Do the same for the depdate<jupyter_code>depdate = us_immigration_df[['depdate']].copy()
# Rename the arrdate column to sas_date for final storage on Redshift
depdate = depdate.rename(columns={"depdate": "sas_date"})

# Make new column so that it provides date in readable format
depdate['date'] = depdate['sas_date'].apply(lambda col: datetime.date(1960, 1, 1) + datetime.timedelta(days=col))

# Also put in the weekday, day, month and year as new columns
depdate['weekday'] = depdate['date'].apply(lambda col: col.weekday())
depdate['day'] = depdate['date'].apply(lambda col: col.day)
depdate['month'] = depdate['date'].apply(lambda col: col.month)
depdate['year'] = depdate['date'].apply(lambda col: col.year)

# Drop duplicates
depdate = depdate.drop_duplicates()
print(depdate.shape)
depdate.head()<jupyter_output>(218, 6)<jupyter_text>Now that we have both arrival and departure dates, we should combine these into one final table as both the arrival and departure dates are both events we are interested in.  Simply do an outer join to help us accomplish this.<jupyter_code>date = pd.merge(arrdate, depdate, how='outer')
print(date.shape)
date.head()<jupyter_output>(218, 6)<jupyter_text>Moving the data onto RedshiftNow that we've prepared our data seen above, let's move it onto Amazon Redshift.  Take note that this will use a configuration file that requires information that is sensitive, so on submission these are nulled out.<jupyter_code>import configparser
# Read in attributes from custom config file to access S3 and Redshift
config = configparser.ConfigParser()
config.read('dwh.cfg')
# Attributes for accessing Redshift cluster
DWH_DB_USER = config.get("CLUSTER", "DB_USER")
DWH_DB_PASSWORD = config.get("CLUSTER", "DB_PASSWORD")
DWH_ENDPOINT = config.get("CLUSTER", "HOST")
DWH_PORT = config.get("CLUSTER", "DB_PORT")
DWH_DB = config.get("CLUSTER", "DB_NAME")
# Attributes for S3 access
AWS_ACCESS_KEY = config.get("AWS", "AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = config.get("AWS", "AWS_SECRET_ACCESS_KEY_ID")
AWS_BUCKET = config.get("AWS", "AWS_BUCKET")
AWS_BUCKET_SUBDIR = config.get("AWS", "AWS_BUCKET_SUBDIR")
# Connect to Redshift cluster
pr.connect_to_redshift(dbname=DWH_DB, host=DWH_ENDPOINT, port=DWH_PORT, user=DWH_DB_USER, password=<PASSWORD>)
# Connect to S3
pr.connect_to_s3(aws_access_key_id=AWS_ACCESS_KEY, aws_secret_access_key=AWS_SECRET_ACCESS_KEY, bucket=AWS_BUCKET, subdirectory=AWS_BUCKET_SUBDIR)<jupyter_output><empty_output><jupyter_text>`pandas_redshift.pandas_to_redshift` is a convenience method that will place our dataframe in an intermediate CSV form on a S3 bucket of our choice, then it will create a new table on Redshift based on this intermediate form.  Think of this as creating a staging table then final table in one sweep.  We will now place our tables on S3, then Redshift.<jupyter_code># Create US immigration fact table - Put on S3 then go to Redshift
pr.pandas_to_redshift(data_frame=us_immigration_df, redshift_table_name = 'public.immigration', index=False)
# Create state dimension table - Put on S3 then go to Redshift
pr.pandas_to_redshift(data_frame=state_pop_df, redshift_table_name = 'public.state')
# Create date dimension table - Put on S3 then go to Redshift
pr.pandas_to_redshift(data_frame=date, redshift_table_name = 'public.date')<jupyter_output>Pandas Redshift | 2021-01-01 16:31:10,697 | pandas_redshift.core | INFO | saved file public.date-08be7eba-3d8c-4f3c-961d-8c70ff5ebf77.csv in bucket capstone_data/public.date-08be7eba-3d8c-4f3c-961d-8c70ff5ebf77.csv
Pandas Redshift | 2021-01-01 16:31:10,702 | pandas_redshift.core | INFO | create table public.date (sas_date REAL, date VARCHAR(256), weekday BIGINT, day BIGINT, month BIGINT, year BIGINT) diststyle even
Pandas Redshift | 2021-01-01 16:31:10,706 | pandas_redshift.core | INFO | CREATING A TABLE IN REDSHIFT
Pandas Redshift | 2021-01-01 16:31:10,949 | pandas_redshift.core | INFO | 
    copy public.date
    from 's3://rayryeng-dend-capstone/capstone_data/public.date-08be7eba-3d8c-4f3c-961d-8c70ff5ebf77.csv'
    delimiter ','
    ignoreheader 1
    csv quote as '"'
    dateformat 'auto'
    timeformat 'auto'
    
        access_key_id '********'
        secret_access_key '********'
        
    
    ;
Pandas Redshift | 2021-01-01 16:31:10,953 | pandas_redshift.core | INFO | FILLI[...]<jupyter_text>4.2 Data Quality ChecksWe will perform integrity checks to ensure that our data is in a sane format.  This will include some sample queries made directly to the tables as well as checking the number of records for each table.  `pandas_redshift.redshift_to_pandas` is a nice function that allows us to provide a SQL query as a string and it returns a Pandas dataframe of the resulting query. 4.2.1 Number of rows should be non-zeroCheck to see if we have any tables that are empty<jupyter_code>def data_quality_check(table):
    """
    Checks to see if the total number of records is non-zero for a table
    
    Inputs:
        table - String to denote our table to check
    
    Returns:
        True/False to signify if number of records is non-zero
    """
    return len(pr.redshift_to_pandas(f"SELECT COUNT(*) FROM {table}")) > 0
for table in ['immigration', 'state', 'date']:
    if not data_quality_check(table):
        print(f"Data quality check failed. {table} returned no results.")
    else:
        print(f"Data quality check passed for {table}")<jupyter_output>Data quality check passed for immigration
Data quality check passed for state
Data quality check passed for date<jupyter_text>4.2.2 Check to see if there are any null primary keysLet's also check to see if there are any null records for the primary keys for the columns.<jupyter_code>def data_quality_check_null(table, column):
    """
    Checks to see if there are any null records in a column
    
    Inputs:
        table - String to denote our table to check
        column - Column to check
    
    Returns:
        True/False to signify if there are null records in a column
    """
    return len(pr.redshift_to_pandas(f"SELECT COUNT(*) FROM {table} WHERE {column} IS NULL")) == 0
for table, column in zip(['immigration', 'state', 'date'], ['cicid', 'state_code', 'sas_date']):
    if data_quality_check_null(table, column):
        print(f"Data quality check failed. {table} has null records for column {column}")
    else:
        print(f"Data quality check passed for {table}, column {column}")<jupyter_output>Data quality check passed for immigration, column cicid
Data quality check passed for state, column state_code
Data quality check passed for date, column sas_date<jupyter_text>4.2.3 Check the first 20 records of our final tablesLet's also do some simple checks on the actual contents.  Cycle through each table, grab 20 records and show them.<jupyter_code>results = {table: pr.redshift_to_pandas(f"SELECT * FROM {table} LIMIT 20") for table in ['immigration', 'state', 'date']}
results['immigration']
results['state']
results['date']
# Finally close the cursor, commit and close the database connection, and remove variables from the environment.
pr.close_up_shop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08458450200736953, 'fraction_numerical': 0.021393609415388, 'mean_word_length': 5.283344851416724, 'pattern_counts': {'":': 2, '<': 60, '<?xml version=': 0, '>': 61, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '493630', 'license': 'MIT', 'n_tokens_mistral': 5698, 'n_tokens_neox': 5349, 'n_words': 2183, 'path': 'capstone-i94-etl-pipeline/Capstone_Project_Final.ipynb', 'repo_name': 'rayryeng/Udacity_Data_Engineering_Nanodegree', 'size': '18188'}