<jupyter_start><jupyter_text>US Immigration Analysis Project SummaryThe goal of this project is to analyze the distribution of the immigrants arrive at US.  Scope For the project use the following tools:- Spark and pandas for data handling- Matplotlib for vizualization DatasetsThe project use the following datasets- **I94 Immigration Data**: This dataset 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). - **U.S. City Demographic Data**: This dataset contains information about the demographics of all US cities and census-designated places with a population greater or equal to 65,000. - **Airport Code Table**: 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).- **Dictionaries**: Contains the lables for the ids in the U.S. City Demographic Data.<jupyter_code># Load librarie
import pyspark
from pyspark.sql.functions import expr, isnan, when, count, col

# Initiate Spark Session
spark = pyspark\
        .sql\
        .SparkSession\
        .builder\
        .config("spark.jars.packages","saurfang:spark-sas7bdat:2.0.0-s_2.11")\
        .enableHiveSupport()\
        .getOrCreate()

# Read parquet data
imm = spark.read.parquet("data_source/sas_data")

# Read csv data
air = spark.read.csv("data_source/airport-codes_csv.csv", header=True)
cit = spark.read.csv("data_source/us-cities-demographics.csv", sep=";", header=True)

# Read dictionaries
i94addr = spark.read.csv("data_source/i94addr.csv", sep=";", header=True)
i94mode = spark.read.csv("data_source/i94mode.csv", sep=";", header=True)
i94port = spark.read.csv("data_source/i94port.csv", sep=";", header=True)
i94visa = spark.read.csv("data_source/i94visa.csv", sep=";", header=True)
i94city = spark.read.csv("data_source/i94city.csv", sep=";", header=True)<jupyter_output><empty_output><jupyter_text>Data ExplorationFor the goal of the project, the focus is on the data related to the:- People arriving at US.    - Quantity    - Dates    - Transport Type    - Location    - Visa Type- Information about the cities    - State    - City    - Population<jupyter_code>imm.printSchema()
air.printSchema()
cit.printSchema()<jupyter_output>root
 |-- City: string (nullable = true)
 |-- State: string (nullable = true)
 |-- Median Age: string (nullable = true)
 |-- Male Population: string (nullable = true)
 |-- Female Population: string (nullable = true)
 |-- Total Population: string (nullable = true)
 |-- Number of Veterans: string (nullable = true)
 |-- Foreign-born: string (nullable = true)
 |-- Average Household Size: string (nullable = true)
 |-- State Code: string (nullable = true)
 |-- Race: string (nullable = true)
 |-- Count: string (nullable = true)<jupyter_text>Check for Missing ValuesCheck any missing values in the Immigration table<jupyter_code>imm.select([count(when(isnan(c), c)).alias(c) for c in imm.columns]).show()<jupyter_output>+-----+-----+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+-------+------+------+-------+------+-----+--------+
|cicid|i94yr|i94mon|i94cit|i94res|i94port|arrdate|i94mode|i94addr|depdate|i94bir|i94visa|count|dtadfile|visapost|occup|entdepa|entdepd|entdepu|matflag|biryear|dtaddto|gender|insnum|airline|admnum|fltno|visatype|
+-----+-----+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-------+-------+------+------+-------+------+-----+--------+
|    0|    0|     0|     0|     0|      0|      0|      0|      0|      0|     0|      0|    0|       0|       0|    0|      0|      0|      0|      0|      0|      0|     0|     0|      0|     0|    0|       0|
+-----+-----+------+------+------+-------+-------+-------+-------+-------+------+-------+-----+--------+--------+-----+-------+-------+-------+-------+-[...]<jupyter_text>Remove rows with the values of the column i94addr are not in U.S. state list<jupyter_code>state = ["AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "DC", "FL", "GA", "GU", "HI", "ID", "IL", "IN", "IA", "KS",
         "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS", "MO", "MT", "NC", "ND", "NE", "NV", "NH", "NJ", "NM", "NY",
         "OH", "OK", "OR", "PA", "PR", "RI", "SC", "SD", "TN", "TX", "UT", "VT", "VI", "VA", "WV", "WA", "WI", "WY"]

imm = imm.where(col("i94addr").isin(state))<jupyter_output><empty_output><jupyter_text>Convert date coluns to correct format<jupyter_code># Converting columns
imm = imm.withColumn('arrdate', expr("date_add(to_date('1960-01-01'),arrdate)"))
imm = imm.withColumn('depdate', expr("date_add(to_date('1960-01-01'),depdate)"))<jupyter_output><empty_output><jupyter_text>Data Model Star SchemaFor the project it will be used the star schema, the star schema consists of one or more fact tables referencing any number of dimension tables.Dimension Tables- Aiport Codes Data- US Cities Data- i94addr (dictionary)- i94city (dictionary)- i94mode (dictionary)- i94mode (dictionary)- i94visa (dictionary)Fact Tables- Immigration Data  Pipelines Pipeline Immigration DistribuitionThe pipeline below is used for the analysis of the distributions of the immigrants<jupyter_code># Merge the tables
pipe = imm.join(i94addr,  ["i94addr"], how="left")
pipe = pipe.join(i94mode, ["i94mode"], how="left")
pipe = pipe.join(i94port, ["i94port"], how="left")
pipe = pipe.join(i94visa, ["i94visa"], how="left")
pipe = pipe.join(i94city, ["i94cit"], how="left")

# Group by State
pipe = pipe.groupby(["arrdate", "i94addr", "State"]).agg({"count": "count"})

# # Filter to print sample of data
pipe.filter(pipe.arrdate == "2016-04-30").show()<jupyter_output>+----------+-------+-----------------+------------+
|   arrdate|i94addr|            State|count(count)|
+----------+-------+-----------------+------------+
|2016-04-30|     MT|          MONTANA|          65|
|2016-04-30|     NH|    NEW HAMPSHIRE|          93|
|2016-04-30|     SD|        S. DAKOTA|          29|
|2016-04-30|     FL|          FLORIDA|       25852|
|2016-04-30|     MD|         MARYLAND|         884|
|2016-04-30|     KS|           KANSAS|          76|
|2016-04-30|     VA|         VIRGINIA|        1222|
|2016-04-30|     ND|        N. DAKOTA|          55|
|2016-04-30|     NM|       NEW MEXICO|         110|
|2016-04-30|     WV|      W. VIRGINIA|          42|
|2016-04-30|     GA|          GEORGIA|        1570|
|2016-04-30|     DC|DIST. OF COLUMBIA|        1185|
|2016-04-30|     ME|            MAINE|          83|
|2016-04-30|     NJ|       NEW JERSEY|        2999|
|2016-04-30|     IN|          INDIANA|         406|
|2016-04-30|     CA|       CALIFORNIA|       19425|
|2016-04-30|[...]<jupyter_text>4.2 Data Quality Checks Run Quality Checks to garanty the dataframe is not null<jupyter_code># Perform quality checks here
print((pipe.count(), len(pipe.columns)))<jupyter_output>(1620, 4)<jupyter_text>4.3 Data dictionary Below the description to each column of the pipeline- arrdate: correspond to the arrive date of the passenger in U.S.- i94addr: abbreviation state- state: the state were the passenger arrive- count: the number of passenger in the state and date  4.4 VisualizationBelow a sample of visualization for the date 04/30/2016<jupyter_code># Load Libraries
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np

data = pipe.filter(pipe.arrdate == "2016-04-30").select("*").toPandas()

objects = data["i94addr"]
y_pos = np.arange(len(objects))
performance = data["count(count)"]
plt.figure(figsize=(18,8))

plt.bar(y_pos, performance, align='center', alpha=0.5)
plt.xticks(y_pos, objects)
plt.ylabel("Passengers Count per State")
plt.xlabel("State")
plt.title("Count Passengers")

plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.21224038343369792, 'fraction_numerical': 0.044856826840358854, 'mean_word_length': 2.727897388914338, 'pattern_counts': {'":': 1, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '274040', 'license': 'MIT', 'n_tokens_mistral': 2916, 'n_tokens_neox': 2651, 'n_words': 746, 'path': 'Data Engineering Capstone Project/Capstone Project Notebook.ipynb', 'repo_name': 'fernandofsilva/Data_Engineering-Udacity', 'size': '8137'}