<jupyter_start><jupyter_text>Data Engineering Capstone Project Project Summary In this project, I designed a STAR-shape schema of the database system, containing the information of U.S. I94 immigration data, country code representation, destination state demographic summary and visa type definitions and constructed the cloud ETL pipeline in **AWS**. Specifically, data is extracted and preprocessed using **Apache Spark (PySpark)** on a provided workspace, dumped into **S3 bucket** and queried using **Redshift**. The goal of this project is to construct the immigration database system efficiently, such that the the query of needed immigration and geographical information is retrieved quickly for further downstream services like analytics. The project follows the follow 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 pandas as pd
from datetime import datetime
from pyspark.sql.types import *
from pyspark.sql import SparkSession
import configparser
import os
from pyspark.sql.functions import isnan, when, count, col, expr, year, month, date_format
from pyspark.sql import SQLContext<jupyter_output><empty_output><jupyter_text>Step 1: Scope the Project and Gather DataI worked with four dataset specifically. - I94 Immigration: This dataset is provided by Udacity, which contains > 3 million rows and is "big". It contains information about foreigners who have entered the U.S. without their P.I.I.. I'm using this table as the main fact table and connect other dimension tables to it. - US city demographic: This dataset is also proviced by Udacity, and it contains city-level demographic distributions of U.S.. I aggregated it to the level of **state** and connected it with the entering state of I94 records.- Country code: This dataset is extracted from the given I94 label ```.SAS``` file, which contains the corresponding information of the country code used by the I94 immigration dataset.- Visa definition: This dataset is collected from [here](https://www.trade.gov/i-94-arrivals-program), which contains the definitions of different visa types of the I94 immigration dataset.  Step 2: Explore and Assess the Data  Country code data<jupyter_code>with open('I94_SAS_Labels_Descriptions.SAS') as f:
    f = f.read().replace('\t', '')
head = f.index("i94cntyl")
tail = f.find(";", head, len(f))
content = [s.replace("'", "") for s in f[head:tail].split("\n")[1:]]
l = []
for c in content:
    code, country = c.split("=")
    code = int(code.strip())
    country = country.strip()
    l.append((code, country))
df_country = pd.DataFrame.from_records(l, columns=["country_code", "country_name"])
df_country.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 289 entries, 0 to 288
Data columns (total 2 columns):
country_code    289 non-null int64
country_name    289 non-null object
dtypes: int64(1), object(1)
memory usage: 4.6+ KB<jupyter_text>State demographic data<jupyter_code>df_city_demo = pd.read_csv("us-cities-demographics.csv", delimiter=";")
columns = ["State Code","Male Population", "Female Population", "Total Population", "Number of Veterans"]
df_state_demo = df_city_demo[columns].groupby("State Code").sum().astype(int).reset_index()
df_state_demo.columns = ["_".join(c.lower().split(" ")) for c in columns]
df_state_demo.info()
del df_city_demo<jupyter_output><empty_output><jupyter_text>Visa data<jupyter_code>with open("visainfo.txt") as f:
    f = f.read().split("\n")
columns = f[0].split(";")
content = []
for l in f[1:]:
    content.append(l.split(";"))
df_visa = pd.DataFrame(content, columns=columns)
df_visa.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 20 entries, 0 to 19
Data columns (total 2 columns):
visa_type    20 non-null object
visa_def     20 non-null object
dtypes: object(2)
memory usage: 400.0+ bytes<jupyter_text>I94 immigration data<jupyter_code># Option 1: Local laod
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["PATH"] = "/opt/conda/bin:/opt/spark-2.4.3-bin-hadoop2.7/bin:/opt/conda/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/lib/jvm/java-8-openjdk-amd64/bin"
os.environ["SPARK_HOME"] = "/opt/spark-2.4.3-bin-hadoop2.7"
os.environ["HADOOP_HOME"] = "/opt/spark-2.4.3-bin-hadoop2.7"
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df_final = spark.read.load('./sas_data')
field_list = ['cicid', 'i94cit', 'i94res', 'i94port', 'i94mode', 'i94addr', 'gender', 'visatype']
df_final = df_final.select(field_list)
# Option 2: Cloud load
config = configparser.ConfigParser()
config.read_file(open('aws/credentials.cfg'))
output_data = config['AWS']['S3_BUCKET']

# Building a spark session with a connection to AWS S3
spark = SparkSession.builder \
        .config("spark.jars.packages","saurfang:spark-sas7bdat:2.0.0-s_2.11","org.apache.hadoop:hadoop-aws:2.7.2") \
        .config("spark.hadoop.fs.s3a.access.key",config['AWS']['AWS_ACCESS_KEY_ID']) \
        .config("spark.hadoop.fs.s3a.secret.key",config['AWS']['AWS_SECRET_ACCESS_KEY']) \
        .getOrCreate()
month_list = ["jan","feb","mar","apr","may","jun","jul","aug","sep","oct","nov","dec"]
field_list = ['cicid', 'i94cit', 'i94res', 'i94port', 'i94mode', 'i94addr', 'gender', 'visatype']

for month in month_list:
    fname = f'../../data/18-83510-I94-Data-2016/i94_{month}16_sub.sas7bdat'
    df_temp = spark.read.format('com.github.saurfang.sas.spark').load(fname)
    df_temp = df_temp.select(field_list)
   
    if month == 'jan':
        df_final = df_temp
    else:
        df_final = df_final.union(df_temp)
# LEFT ANTI JOIN DUP
df_final = df_final.cache()
df_final_dup = df_final.groupby('cicid').count().filter(col('count')>1)
df_final2 = df_final.join(df_final_dup, on=['cicid'], how='left_anti')
df_final2 = df_final2.cache()
# FILL NA
df_final2 = df_final2.fillna({'i94cit':-1, 'i94mode':-1, 'i94addr':'99'})
# CONVERT TO INTEGER
cast_lst = ['cicid','i94cit','i94res','i94mode']

for col_name in cast_lst:
    df_final2 = df_final2.withColumn(col_name, col(col_name).cast('int'))
# Save to .parquet
output_data = "./data/"
df_final2 = df_final2.coalesce(4)
df_final2.write.parquet(f"{output_data}I94-Data")
print(df_final2.count())
df_final2.printSchema()
sc = spark.sparkContext
sqlCtx = SQLContext(sc)
schema = StructType([ \
    StructField("state_code", StringType(), True), \
    StructField("male_population", IntegerType(), True), \
    StructField("female_population", IntegerType(), True), \
    StructField("total_population", IntegerType(), True), \
    StructField("number_of_veterans", IntegerType(), True) \
  ])

spark_state_demo = sqlCtx.createDataFrame(df_state_demo, schema=schema)
spark_state_demo.write.parquet(f"{output_data}demographics")

print(spark_state_demo.count())
spark_state_demo.printSchema()
schema = StructType([ \
    StructField("country_code", IntegerType(), True), \
    StructField("country_name", StringType(), True) \
  ])

spark_country = sqlCtx.createDataFrame(df_country, schema=schema)
spark_country.write.parquet(f"{output_data}country")

print(spark_country.count())
spark_country.printSchema()
schema = StructType([ \
    StructField("visa_type", StringType(), True), \
    StructField("visa_def", StringType(), True) \
  ])

spark_visa = sqlCtx.createDataFrame(df_visa, schema=schema)
spark_visa.write.parquet(f"{output_data}visa")

print(spark_visa.count())
spark_visa.printSchema()
# Create metadata.json
df_metadata = pd.DataFrame([("fact_immigration", df_final2.count()), 
                            ("dim_country", spark_country.count()), 
                            ("dim_visa", spark_visa.count()), 
                            ("dim_state_demo", spark_state_demo.count())], columns=["table", "rows"])
df_metadata.to_json("./metadata.json")<jupyter_output><empty_output><jupyter_text>Step 3: Define the Data Model 3.1 Conceptual Data ModelData model is designed as follows:- Fact table:    - fact_immigration: comes from the I94 immigration data.- Dimension table:    - dim_country: comes from the I94 label SAS file.    - dim_visa: comes from the collected visa definitions.    - dim_state_demo: comes from the aggregation of city-level demographic information.    ![schema](./DB_schema.svg) 3.2 Mapping Out Data PipelinesThe ETL pipeline is as follows: - Create staging, dimension and fact tables.- Load processed ```.parquet``` files from S3.- Insert the loaded data to the corresponding dimension and fact tables in Redshift.![pipeline](./Pipeline.svg)  Step 4: Run Pipelines to Model the Data  4.1 Create the data modelBuild the data pipelines to create the data model.<jupyter_code># Create Redshift cluster
!python ./operations/create_redshift_cluster.py create
# Create tables
!python ./operations/create_tables.py
# Perform ETL pipeline
!python ./operations/etl.py<jupyter_output><empty_output><jupyter_text>4.2 Data Quality ChecksTwo data quality checks are performed:- whether the created table is in the workspace.- whether the number of records in each table is identical to the preprocessed ones.<jupyter_code># Perform data quality check
!python ./operations/data_quality.py
# Delete (the expensive) Redshift cluster
!python ./operations/create_redshift_cluster.py delete<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13171042840437971, 'fraction_numerical': 0.02041033273094504, 'mean_word_length': 5.165137614678899, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 31, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '622776', 'license': 'MIT', 'n_tokens_mistral': 3165, 'n_tokens_neox': 3009, 'n_words': 825, 'path': 'Capstone Project/Capstone Project Template.ipynb', 'repo_name': 'zl3311/Udacity_Data_Engineering_Nanodegree', 'size': '9407'}