<jupyter_start><jupyter_text>Project 08 - Analysis of U.S. Immigration (I-94) Data Udacity Data Engineer - Capstone Project> by <NAME> | 2021-05-05 Project OverviewThis project works with a data set for immigration to the United States. The supplementary datasets will include data onairport codes, U.S. city demographics and temperature data.The following process is divided into different sub-steps to illustrate how to answer the questions set by the businessanalytics team.The project file follows the following steps:* Step 3: Define the Data Model* Step 4: Run ETL to Model the Data 3.1.3. At what times do foreign persons arrive for immigration to the U.S.? [(Data pipeline)](question3_data_pipeline) **Date dimensions**`st_i94_arrdate` and `st_i94_depdate` from staging table `st_i94_immigration` describe dates in SAS specific Date format.The SAS date calculation starts on 1960-01-01. These columns are converted to DateType format in the staging table`st_i94_immigrations` as columns named `st_i94_arrdate_iso` and `st_i94_arrdate_iso`.Get date values from columns `st_i94_immigration.st_i94_arrdate_iso` and `st_i94_immigration.st_i94_depdate_iso`.Get a valid MIN(), MAX() and default (null value representation) date. Clean data and rewrite staging table 'st_i94_immigrations' if needed.Finally, create two dimensions 'd_date_arrivals' and 'd_date_departures' out of it without gaps.1. Read data and get min() and max() value out of `st_i94_arrdate_iso` and `st_i94_depdate_iso`2. Clean date column "st_i94_depdate_iso": Valid entries are between 2016-01-01 and 2017-06-14. Pre- and descending values   will be set to null / default value (1900-01-01)3. Update fact table `f_i94_immigrations` based on cleaned column `st_i94_depdate_iso`  values inside4. Generate new date staging tables (`st_date_arrivals`, `st_date_departures`) based on default, min and max values5. Append date specific columns to staging tables, create a dimension out of it and store it6. Map dimension `d_date_arrivals` to  fact table `f_i94_immigration` based on columns   (`st_date_arrivals.st_da_date` --> `d_date_arrivals.d_da_id`) == (`st_i94_immigration.st_i94_arrdate_iso` --> `f_i94_immigration.d_da_id`).7. Map dimension `d_date_departures` to  fact table `f_i94_immigration` based on columns   (`st_date_departures.st_dd_date` --> `d_date_departures.d_dd_id`) == (`st_i94_immigration.st_i94_depdate_iso` --> `f_i94_immigration.d_dd_id`).8. Answer Project Question 3.1: At what times do foreign persons arrive for immigration to the U.S.?9. Answer Project Question 3.2: When a foreign person comes to the U.S. for immigration, do they travel on to another state?10. Answer Project Question 3.3: If a foreign person travels to another state, after which period of time does this happen?The creation of those two date dimensions is based on one physical table. This method is called[Role-Playing Dimensions](https://dba.stackexchange.com/questions/137971/how-many-date-dimensions-for-one-fact)![Role-Playing Dimension](../P8_capstone_documentation/11_P8_RolePlayingDimension.png). Step 4: Run ETL to Model the Data 4.1.3. At what times do foreign persons arrive for immigration to the U.S.? [(Data description)](question3_description) **Date dimensions**`st_i94_arrdate` and `st_i94_depdate` from staging table `st_i94_immigration` describe dates in SAS specific Date format.The SAS date calculation starts on 1960-01-01. These columns are converted to DateType format in the staging table`st_i94_immigrations` as columns named `st_i94_arrdate_iso` and `st_i94_arrdate_iso`. Get date values from columns `st_i94_immigration.st_i94_arrdate_iso` and `st_i94_immigration.st_i94_depdate_iso`.Get a valid MIN(), MAX() and default (null value representation) date. Clean data and rewrite staging table 'st_i94_immigrations' if needed.Finally, create two dimensions 'd_date_arrivals' and 'd_date_departures' out of it without gaps.1. Read data and get min() and max() value out of `st_i94_arrdate_iso` and `st_i94_depdate_iso`<jupyter_code>###### Imports and Installs section
import shutil
import pandas as pd
import pyspark.sql.functions as F
# import spark as spark
from pyspark.sql.types import StructType, StructField, DoubleType, StringType, IntegerType, LongType, TimestampType, DateType
from datetime import datetime, timedelta
from pyspark.sql import SparkSession, DataFrameNaFunctions
from pyspark.sql.functions import when, count, col, to_date, datediff, date_format, month
import re
import json
from os import path
MAX_MEMORY = "5g"

spark = SparkSession\
    .builder\
    .appName("etl pipeline for project 8 - I94 data") \
    .config("spark.jars.packages","saurfang:spark-sas7bdat:3.0.0-s_2.12")\
    .config('spark.sql.repl.eagerEval.enabled', True) \
    .config("spark.executor.memory", MAX_MEMORY) \
    .config("spark.driver.memory", MAX_MEMORY) \
    .appName("Foo") \
    .enableHiveSupport()\
    .getOrCreate()

# setting the current LOG-Level
spark.sparkContext.setLogLevel('ERROR')
# Read written data frame back into memory
location_to_read = "../P8_capstone_resource_files/parquet_stage/PQ2/st_i94_immigrations"
df_st_i94_immigrations = spark.read.parquet(location_to_read)
df_st_i94_immigrations.printSchema()
# Get an overview about valid data - check some different perspectives
# get valid min and max date from date_fields `st_i94_arrdate_iso` and `st_i94_depdate_iso`

st_i94_arrdate_depdate_iso = df_st_i94_immigrations.select(
    F.min(col("st_i94_arrdate_iso")).alias("st_i94_arrdate_iso_min"),
    F.max(col("st_i94_arrdate_iso")).alias("st_i94_arrdate_iso_max"),
    F.min(col("st_i94_depdate_iso")).alias("st_i94_depdate_iso_min"),
    F.max(col("st_i94_depdate_iso")).alias("st_i94_depdate_iso_max"),
)

print(st_i94_arrdate_depdate_iso)

# get an overview about arrdate:
# all distinct date values
print(df_st_i94_immigrations.select("st_i94_arrdate_iso").distinct().count())
# Most entries on which date?
df_st_i94_immigrations.select("st_i94_arrdate_iso").groupBy("st_i94_arrdate_iso").count().sort("count", ascending=False).show(1000, False)
# get all date values. Is there a large gap or date values out of range?
df_st_i94_immigrations.select("st_i94_arrdate_iso").groupBy("st_i94_arrdate_iso").count().sort("st_i94_arrdate_iso", ascending=True).show(1000, False)

"""
Findings:

Everything seems to be valid. Date values start from 1st of January 2016 and ends by 31st of December 2016.
"""
# get an overview about depdate:
# all distinct date values
print(df_st_i94_immigrations.select("st_i94_depdate_iso").distinct().count())
# Most entries on which date?
df_st_i94_immigrations.select("st_i94_depdate_iso").groupBy("st_i94_depdate_iso").count().sort("count", ascending=False).show(10, False)
# get all date values. Is there a large gap or date values out of range?
df_st_i94_immigrations.select("st_i94_depdate_iso").groupBy("st_i94_depdate_iso").count().sort("st_i94_depdate_iso", ascending=True).show(1000, False)
# compare st_i94_arrdate with st_i94_depdate. Is departure date earlier than arrival date --> there is a logical failure!

# Show only data to be corrected in the third column
df_st_i94_immigrations \
    .groupBy("st_i94_arrdate_iso", "st_i94_arrdate", "st_i94_depdate", "st_i94_depdate_iso") \
    .count() \
    .withColumn("st_i94_depdate_iso_wrong_dates",
                 when(col("st_i94_depdate_iso") < "2016-01-01", "1111-01-01")\
                .when(col("st_i94_depdate_iso") > "2017-06-14", "2222-01-01") \
                .when(col("st_i94_arrdate_iso") > col("st_i94_depdate_iso"), "3333-01-01")
                .otherwise(" ").cast(StringType())) \
    .orderBy("st_i94_arrdate_iso", "st_i94_depdate_iso") \
    .show(20)

"""
Findings:
715 different date values ==> all distinct date values are greater than (>) 366 days of a year
==> That's possible. Many Immigrants already know their departure date.

1900-01-01 (start date):  This date is used as default value instead of a null value

arrdate starts on 2016-01-01 ==> The departure date cannot be earlier than the arrival date! --> each date before
2016-01-01 must be set to 1900-01-01 as null/default value

depdate greater than 2017-06-14 is not realistic, due to the very small amount of depdate entries within this range of dates
==> entries must be set to 1900-01-01 (null/default)

arrdate describes the 1st arrival into the U.S.. After that the immigrants decide to travel to different states in the U.S..
conclusion: arrdate must be earlier than depdate (arrdate < depdate ==> 2016-01-01 < 2016-01-02)

The following table shows some wrong dates where arrdate > depdate
+------------------+--------------+--------------+------------------+-----+------------------------------+
|st_i94_arrdate_iso|st_i94_arrdate|st_i94_depdate|st_i94_depdate_iso|count|st_i94_depdate_iso_wrong_dates|
+------------------+--------------+--------------+------------------+-----+------------------------------+
|        2016-01-02|         20455|         20454|        2016-01-01|    1|                    3333-01-01|
|        2016-01-08|         20461|         20454|        2016-01-01|    1|                    3333-01-01|
|        2016-01-08|         20461|         20459|        2016-01-06|    2|                    3333-01-01|
|        2016-01-08|         20461|         20460|        2016-01-07|    3|                    3333-01-01|
+------------------+--------------+--------------+------------------+-----+------------------------------+
"""<jupyter_output><empty_output><jupyter_text>2. Clean date column "st_i94_depdate_iso" and "st_": Valid entries are between 2016-01-01 and 2017-06-14. Pre- and   descending values will be set to null / default value (1900-01-01)<jupyter_code># show corrected column `st_i94_depdate_iso_corrected`
df_st_i94_immigrations \
    .groupBy("st_i94_arrdate_iso", "st_i94_arrdate", "st_i94_depdate", "st_i94_depdate_iso") \
    .count() \
    .withColumn("st_i94_depdate_iso_corrected",
                 when(col("st_i94_depdate_iso") < "2016-01-01", "1900-01-01")\
                .when(col("st_i94_depdate_iso") > "2017-06-14", "1900-01-01") \
                .when(col("st_i94_arrdate_iso") > col("st_i94_depdate_iso"), "1900-01-01")
                .otherwise(col("st_i94_depdate_iso")).cast(DateType())) \
    .orderBy("st_i94_arrdate_iso", "st_i94_depdate_iso") \
    .show(20)
# correct the date values in column `st_i94_depdate_iso`
df_st_i94_immigrations = df_st_i94_immigrations \
    .withColumn("st_i94_depdate_iso",
                 when(col("st_i94_depdate_iso") < "2016-01-01", "1900-01-01") \
                .when(col("st_i94_depdate_iso") > "2017-06-14", "1900-01-01") \
                .when(col("st_i94_arrdate_iso") > col("st_i94_depdate_iso"), "1900-01-01")
                .otherwise(col("st_i94_depdate_iso")).cast(DateType()))
df_st_i94_immigrations \
    .groupBy("st_i94_arrdate_iso", "st_i94_arrdate", "st_i94_depdate", "st_i94_depdate_iso") \
    .count() \
    .withColumn("st_i94_depdate_iso_wrong_dates",
                 when(col("st_i94_depdate_iso") < "2016-01-01", "1111-01-01")\
                .when(col("st_i94_depdate_iso") > "2017-06-14", "2222-01-01") \
                .when(col("st_i94_arrdate_iso") > col("st_i94_depdate_iso"), "3333-01-01")
                .otherwise(" ").cast(StringType())) \
    .orderBy("st_i94_depdate_iso_wrong_dates", ascending=False) \
    .show(20)
df_st_i94_immigrations.printSchema()
df_st_i94_immigrations.show(50)
# write st_i94_immigrations back to file system
location_to_write = "../P8_capstone_resource_files/parquet_stage/PQ3/st_i94_immigrations"

# delete folder if already exists
if path.exists(location_to_write):
    shutil.rmtree(location_to_write)

df_st_i94_immigrations \
    .repartition(int(1)) \
    .write \
    .format("parquet")\
    .mode(saveMode='overwrite') \
    .partitionBy('st_i94_year', 'st_i94_month') \
    .parquet(location_to_write, compression="gzip")<jupyter_output><empty_output><jupyter_text>3. Update fact table `f_i94_immigrations` based on cleaned column `st_i94_depdate_iso`  values inside<jupyter_code># Read data frames back into memory
# st_i94_immigrations with column `st_i94_port_state_code`:
location_st_i94_immigrations = "../P8_capstone_resource_files/parquet_stage/PQ3/st_i94_immigrations"
df_st_i94_immigrations = spark.read.parquet(location_st_i94_immigrations)

# f_i94_immigrations:
location_f_i94_immigrations = "../P8_capstone_resource_files/parquet_star/PQ2/f_i94_immigrations"
df_f_i94_immigrations = spark.read.parquet(location_f_i94_immigrations)

# show current schemas
print(df_st_i94_immigrations.count())
df_st_i94_immigrations.printSchema()
df_st_i94_immigrations.show(5,False)

print(df_f_i94_immigrations.count())
df_f_i94_immigrations.printSchema()
df_f_i94_immigrations.show(5,False)
# add column 'st_i94_depdate_iso' to fact table 'f_i94_immigrations'
df_st_i94_immigrations_2_join = df_st_i94_immigrations \
    .select("st_i94_id" , "st_i94_depdate_iso")

df_st_i94_immigrations_2_join.select("st_i94_id").printSchema()
df_f_i94_immigrations.select("f_i94_id").printSchema()
df_st_i94_immigrations_2_join.select("st_i94_id").filter("st_i94_id == 8987702").show()
df_f_i94_immigrations.select("f_i94_id").filter("f_i94_id == 8987702").show(1)
"""
Check that there are no null values in columns to join! Otherwise you can run out of memory!
A hint came from

https://medium.com/@yhoso/resolving-weird-spark-errors-f34324943e1c
-->An error occurred while calling o64.cacheTable. or An error occurred while calling o206.showString
"""
# check whether there are still zero values in the result data frame
df_st_i94_immigrations_2_join\
    .select([count( when(col(c).isNull(), c) )
            .alias(c) for c in df_st_i94_immigrations_2_join.columns])\
    .toPandas().T
df_f_i94_immigrations\
    .select([count( when(col(c).isNull(), c) )
            .alias(c) for c in df_f_i94_immigrations.columns])\
    .toPandas().T
# fill up null values with NA
df_f_i94_immigrations = df_f_i94_immigrations.fillna(value='NA', subset=['f_i94_port_state_code'])
df_f_i94_immigrations\
    .select([count( when(col(c).isNull(), c) )
            .alias(c) for c in df_f_i94_immigrations.columns])\
    .toPandas().T
# .coalesce(5) --> split df in partitions
df_f_i94_immigrations = df_f_i94_immigrations \
    .coalesce(5) \
    .join(df_st_i94_immigrations_2_join, F.col("st_i94_id") == F.col("f_i94_id")
          , 'inner') \
    .withColumn("f_i94_depdate_iso", col("st_i94_depdate_iso")) \
    .drop("st_i94_id", "st_i94_depdate_iso")
# update column 'd_dd_id' with newly updated values from column 'f_i94_depdate_iso'
df_f_i94_immigrations = df_f_i94_immigrations \
    .withColumn("d_dd_id", col("f_i94_depdate_iso"))
# check again if values of referencing column "d_dd_id" are equal to column "f_i94_depdate_iso"
df_f_i94_immigrations \
    .filter(col("d_dd_id") != col("f_i94_depdate_iso")) \
    .groupBy("d_dd_id", "f_i94_depdate_iso") \
    .count() \
    .orderBy("d_dd_id") \
    .show(5)
# write st_i94_immigrations back to file system
location_to_write = "../P8_capstone_resource_files/parquet_star/PQ3/f_i94_immigrations"

# delete folder if already exists
if path.exists(location_to_write):
    shutil.rmtree(location_to_write)

df_f_i94_immigrations \
    .repartition(1) \
    .write \
    .format("parquet")\
    .mode(saveMode='overwrite')\
    .partitionBy('f_i94_year', 'f_i94_month') \
    .parquet(location_to_write, compression="gzip")<jupyter_output><empty_output><jupyter_text>4. Generate new date staging tables (`st_date_arrivals`, `st_date_departures`) based on default, min and max values<jupyter_code># Create new data frame with date series
def generate_dates(spark,range_list, dt_col="date_time_ref", interval=60*60*24): # TODO: attention to sparkSession
    """
    ...     Create a Spark DataFrame with a single column named dt_col and a range of date within a specified interval
            (start and stop included).
    ...     With hourly data, dates end at 23 of stop day
    ...     (https://stackoverflow.com/questions/57537760/pyspark-how-to-generate-a-dataframe-composed-of-datetime-range)
    ...
    ...     :param spark: SparkSession or sqlContext depending on environment (server vs local)
    ...     :param range_list: array of strings formatted as "2018-01-20" or "2018-01-20 00:00:00"
    ...     :param interval: number of seconds (frequency), output from get_freq()
    ...     :param dt_col: string with date column name. Date column must be TimestampType
    ...
    ...     :returns: df from range
    ...     """
    start,stop = range_list
    temp_df = spark.createDataFrame([(start, stop)], ("start", "stop"))
    temp_df = temp_df.select([F.col(c).cast("timestamp") for c in ("start", "stop")])
    temp_df = temp_df.withColumn("stop",F.date_add("stop",1).cast("timestamp"))
    temp_df = temp_df.select([F.col(c).cast("long") for c in ("start", "stop")])
    start, stop = temp_df.first()
    return spark.range(start,stop,interval).select(F.col("id").cast("timestamp").cast("date").alias(dt_col))
# Create new staging tables 'st_date_arrivals' and 'st_date_departure' with min and max date values
location_to_read = "../P8_capstone_resource_files/parquet_star/PQ3/f_i94_immigrations"
df_f_i94_immigrations = spark.read.parquet(location_to_read)

print(df_f_i94_immigrations.count())
df_f_i94_immigrations.printSchema()
# check if all date values from "f_i94_arrdate_iso" are valid
df_f_i94_immigrations\
    .groupBy("f_i94_arrdate_iso")\
    .count()\
    .orderBy("f_i94_arrdate_iso")\
    .show(5000)
# Get min and max values for "f_i94_arrdate"
f_i94_arrdate_iso_min, f_i94_arrdate_iso_max =  df_f_i94_immigrations \
    .select(F.min("f_i94_arrdate_iso").alias("f_i94_arrdate_iso_min"), \
            F.max("f_i94_arrdate_iso").alias("f_i94_arrdate_iso_max")) \
    .first()


print(f"f_i94_arrdate_iso_min: {f_i94_arrdate_iso_min}")
print(f"f_i94_arrdate_iso_max: {f_i94_arrdate_iso_max}")

# create new staging table "st_date_arrivals"
date_range = [f_i94_arrdate_iso_min, f_i94_arrdate_iso_max]
dt_col="st_da_date"
df_st_date_arrivals = generate_dates(spark, date_range, dt_col)

df_st_date_arrivals.printSchema()
df_st_date_arrivals.head(5)
df_st_date_arrivals.tail(5)<jupyter_output><empty_output><jupyter_text>5. Append date specific columns to staging tables, create a dimension from it and save it to the file system.<jupyter_code># create new columns of st_date_arrivals table
# https://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html

df_st_date_arrivals = df_st_date_arrivals \
    .withColumn("st_da_id", col("st_da_date")) \
    .withColumn("st_da_year", F.year(col("st_da_date"))) \
    .withColumn("st_da_year_quarter", F.concat_ws('/', F.year(col("st_da_date")), F.quarter(col("st_da_date")))) \
    .withColumn("st_da_year_month", F.concat_ws('/', F.year(col("st_da_date")), F.month(col("st_da_date")))) \
    .withColumn("st_da_year_month", F.concat_ws('/', F.year(col("st_da_date")), date_format(col("st_da_date"), 'MM'))) \
    .withColumn("st_da_quarter", F.quarter(col("st_da_date"))) \
    .withColumn("st_da_month", F.month(col("st_da_date"))) \
    .withColumn("st_da_week", F.weekofyear(col("st_da_date"))) \
    .withColumn("st_da_weekday", F.date_format(col("st_da_date"),'EEEE')) \
    .withColumn("st_da_weekday_short", F.date_format(col("st_da_date"),'EEE')) \
    .withColumn("st_da_dayofweek", F.dayofweek(col("st_da_date"))) \
    .withColumn("st_da_day", F.dayofmonth(col("st_da_date")) )

df_st_date_arrivals.printSchema()
df_st_date_arrivals.show(5)
# persist staging time table 'st_date_arrivals'
location_to_write = "../P8_capstone_resource_files/parquet_stage/PQ3/st_date_arrivals"

# delete folder if already exists
if path.exists(location_to_write):
    shutil.rmtree(location_to_write)

df_st_date_arrivals \
    .repartition(int(1)) \
    .write \
    .format("parquet")\
    .mode(saveMode='overwrite') \
    .parquet(location_to_write, compression="gzip")
# create dimension 'd_date_arrivals' from staging table 'st_date_arrivals'
location_to_read = "../P8_capstone_resource_files/parquet_stage/PQ3/st_date_arrivals"
df_st_date_arrivals = spark.read.parquet(location_to_read)

print(df_st_date_arrivals.count())
df_st_date_arrivals.printSchema()

df_st_date_arrivals = df_st_date_arrivals \
    .withColumnRenamed("st_da_date", "d_da_date") \
    .withColumnRenamed("st_da_id", "d_da_id") \
    .withColumnRenamed("st_da_year", "d_da_year") \
    .withColumnRenamed("st_da_year_quarter", "d_da_year_quarter") \
    .withColumnRenamed("st_da_year_month", "d_da_year_month") \
    .withColumnRenamed("st_da_quarter", "d_da_quarter") \
    .withColumnRenamed("st_da_month", "d_da_month") \
    .withColumnRenamed("st_da_week", "d_da_week") \
    .withColumnRenamed("st_da_weekday", "d_da_weekday") \
    .withColumnRenamed("st_da_weekday_short", "d_da_weekday_short") \
    .withColumnRenamed("st_da_dayofweek", "d_da_dayofweek") \
    .withColumnRenamed("st_da_day", "d_da_day") \

df_st_date_arrivals.printSchema()
df_st_date_arrivals.show(5)


location_to_write = "../P8_capstone_resource_files/parquet_star/PQ3/d_date_arrivals"

# delete folder if already exists
if path.exists(location_to_write):
    shutil.rmtree(location_to_write)

df_st_date_arrivals \
    .repartition(int(1)) \
    .write \
    .format("parquet")\
    .mode(saveMode='overwrite') \
    .parquet(location_to_write, compression="gzip")
# Creation of the second dimension named `d_date_departures` based on fact column `f_i94_depdate_iso`.
# Create new staging table 'st_date_departure' with min, max and default date values
location_to_read = "../P8_capstone_resource_files/parquet_star/PQ3/f_i94_immigrations"
df_f_i94_immigrations = spark.read.parquet(location_to_read)

print(df_f_i94_immigrations.count())
df_f_i94_immigrations.printSchema()
# check if all date values from "f_i94_depdate_iso" are valid
df_f_i94_immigrations\
    .groupBy("f_i94_depdate_iso")\
    .count()\
    .orderBy("f_i94_depdate_iso")\
    .show(5000)
# extract default, min and max date from column 'f_i94_depdate_iso'
# get default and min value
f_i94_depdate_iso_default, f_i94_depdate_iso_min = df_f_i94_immigrations\
    .select("f_i94_depdate_iso") \
    .distinct() \
    .orderBy("f_i94_depdate_iso", ascending=True) \
    .limit(2) \
    .select(F.min("f_i94_depdate_iso").alias("f_i94_depdate_iso_default"),
            F.max("f_i94_depdate_iso").alias("f_i94_depdate_iso_min")) \
    .first()

# get max value
f_i94_depdate_iso_max, f_i94_depdate_iso_max =  df_f_i94_immigrations \
    .select(F.max("f_i94_depdate_iso").alias("f_i94_depdate_iso_max"), \
            F.max("f_i94_depdate_iso").alias("f_i94_depdate_iso_max")) \
    .first()

# check selected data
print(f"f_i94_depdate_iso_default: {f_i94_depdate_iso_default}")
print(f"f_i94_depdate_iso_min: {f_i94_depdate_iso_min}")
print(f"f_i94_depdate_iso_max: {f_i94_depdate_iso_max}")
# create new staging table "st_date_departures"
date_range_default = [f_i94_depdate_iso_default, f_i94_depdate_iso_default]
date_range_min_max = [f_i94_depdate_iso_min, f_i94_depdate_iso_max]

# check valid date ranges
print(date_range_default)
print(date_range_min_max)

# create new data frames for
dt_col="st_dd_date"
df_st_date_departures_default = generate_dates(spark, date_range_default, dt_col)
df_st_date_departures_min_max = generate_dates(spark, date_range_min_max, dt_col)
# combine both data frames to append `1900-01-01` to all other dates
df_st_date_departures = df_st_date_departures_default.union(df_st_date_departures_min_max)
df_st_date_departures.printSchema()
df_st_date_departures.head(5)
df_st_date_departures.tail(5)


# Append date specific columns to staging table `st_date_departures`.
# create new columns of st_date_departures table
# https://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html

df_st_date_departures = df_st_date_departures \
    .withColumn("st_dd_id", col("st_dd_date")) \
    .withColumn("st_dd_year", F.year(col("st_dd_date"))) \
    .withColumn("st_dd_year_quarter", F.concat_ws('/', F.year(col("st_dd_date")), F.quarter(col("st_dd_date")))) \
    .withColumn("st_dd_year_month", F.concat_ws('/', F.year(col("st_dd_date")), date_format(col("st_dd_date"), "MM")) )\
    .withColumn("st_dd_quarter", F.quarter(col("st_dd_date"))) \
    .withColumn("st_dd_month", F.month("st_dd_date")) \
    .withColumn("st_dd_week", F.weekofyear(col("st_dd_date"))) \
    .withColumn("st_dd_weekday", F.date_format(col("st_dd_date"),'EEEE')) \
    .withColumn("st_dd_weekday_short", F.date_format(col("st_dd_date"),'EEE')) \
    .withColumn("st_dd_dayofweek", F.dayofweek(col("st_dd_date"))) \
    .withColumn("st_dd_day", F.dayofmonth(col("st_dd_date")) )
# get prepared staging table
df_st_date_departures.printSchema()
df_st_date_departures.show(5)
# persist staging time table 'st_date_departures'
location_to_write = "../P8_capstone_resource_files/parquet_stage/PQ3/st_date_deaprtures"

# delete folder if already exists
if path.exists(location_to_write):
    shutil.rmtree(location_to_write)

df_st_date_departures \
    .repartition(int(1)) \
    .write \
    .format("parquet")\
    .mode(saveMode='overwrite') \
    .parquet(location_to_write, compression="gzip")
# create dimension 'd_date_arrivals' from staging table 'st_date_arrivals'
location_to_read = "../P8_capstone_resource_files/parquet_stage/PQ3/st_date_deaprtures"
df_st_date_departures = spark.read.parquet(location_to_read)

print(df_st_date_departures.count())
df_st_date_departures.printSchema()


df_st_date_departures = df_st_date_departures \
    .withColumnRenamed("st_dd_date", "d_dd_date") \
    .withColumnRenamed("st_dd_id", "d_dd_id") \
    .withColumnRenamed("st_dd_year", "d_dd_year") \
    .withColumnRenamed("st_dd_year_quarter", "d_dd_year_quarter") \
    .withColumnRenamed("st_dd_year_month", "d_dd_year_month") \
    .withColumnRenamed("st_dd_quarter", "d_dd_quarter") \
    .withColumnRenamed("st_dd_month", "d_dd_month") \
    .withColumnRenamed("st_dd_week", "d_dd_week") \
    .withColumnRenamed("st_dd_weekday", "d_dd_weekday") \
    .withColumnRenamed("st_dd_weekday_short", "d_dd_weekday_short") \
    .withColumnRenamed("st_dd_dayofweek", "d_dd_dayofweek") \
    .withColumnRenamed("st_dd_day", "d_dd_day") \

df_st_date_departures.printSchema()
df_st_date_departures.show(5)


location_to_write = "../P8_capstone_resource_files/parquet_star/PQ3/d_date_departures"

# delete folder if already exists
if path.exists(location_to_write):
    shutil.rmtree(location_to_write)

df_st_date_departures \
    .repartition(int(1)) \
    .write \
    .format("parquet")\
    .mode(saveMode='overwrite') \
    .parquet(location_to_write, compression="gzip")<jupyter_output>531
root
 |-- st_dd_date: date (nullable = true)
 |-- st_dd_id: date (nullable = true)
 |-- st_dd_year: integer (nullable = true)
 |-- st_dd_year_quarter: string (nullable = true)
 |-- st_dd_year_month: string (nullable = true)
 |-- st_dd_quarter: integer (nullable = true)
 |-- st_dd_month: integer (nullable = true)
 |-- st_dd_week: integer (nullable = true)
 |-- st_dd_weekday: string (nullable = true)
 |-- st_dd_weekday_short: string (nullable = true)
 |-- st_dd_dayofweek: integer (nullable = true)
 |-- st_dd_day: integer (nullable = true)

root
 |-- d_dd_date: date (nullable = true)
 |-- d_dd_id: date (nullable = true)
 |-- d_dd_year: integer (nullable = true)
 |-- d_dd_year_quarter: string (nullable = true)
 |-- d_dd_year_month: string (nullable = true)
 |-- d_dd_quarter: integer (nullable = true)
 |-- d_dd_month: integer (nullable = true)
 |-- d_dd_week: integer (nullable = true)
 |-- d_dd_weekday: string (nullable = true)
 |-- d_dd_weekday_short: string (nullable = true)
 |-- d_dd[...]<jupyter_text>6. Map dimension `d_date_arrivals` to  fact table `f_i94_immigration` based on columns   (`st_date_arrivals.st_da_date` --> `d_date_arrivals.d_da_id`) == (`st_i94_immigration.st_i94_arrdate_iso` -->   `f_i94_immigration.d_da_id`).7. Map dimension `d_date_departures` to  fact table `f_i94_immigration` based on columns   (`st_date_departures.st_dd_date` --> `d_date_departures.d_dd_id`) == (`st_i94_immigration.st_i94_depdate_iso` -->   `f_i94_immigration.d_dd_id`).8. Answer Project Question 3: At what times do foreign persons arrive for immigration to the U.S.?<jupyter_code># reload fact table
location_to_read = "../P8_capstone_resource_files/parquet_star/PQ3/f_i94_immigrations"
df_f_i94_immigrations = spark.read.parquet(location_to_read)
print(df_f_i94_immigrations.count())
df_f_i94_immigrations.printSchema()
df_f_i94_immigrations.show(5, False)

location_to_read = "../P8_capstone_resource_files/parquet_star/PQ3/d_date_arrivals"
df_d_date_arrivals = spark.read.parquet(location_to_read)
print(df_d_date_arrivals.count())
df_d_date_arrivals.printSchema()
df_d_date_arrivals.show(5, False)

location_to_read = "../P8_capstone_resource_files/parquet_star/PQ3/d_date_departures"
df_d_date_departures = spark.read.parquet(location_to_read)
print(df_d_date_departures.count())
df_d_date_departures.printSchema()
df_d_date_departures.show(5, False)
# Register data frames as Views
df_f_i94_immigrations.createOrReplaceTempView("f_i94_immigrations")
df_d_date_arrivals.createOrReplaceTempView("d_date_arrivals")
df_d_date_departures.createOrReplaceTempView("d_date_departures")<jupyter_output><empty_output><jupyter_text>8. Answer Project Question 3.1: At what times do foreign persons arrive for immigration to the U.S.?<jupyter_code># SQL to answer Project Question 3.1: At what times do foreign persons arrive for immigration to the U.S.?
df_pq3_1 = spark.sql("select da.d_da_year_month as Year_Month"
                     "      ,count(f_i94.f_i94_count) as  Immigrants"
                     "      ,RANK() OVER (ORDER BY count(f_i94.f_i94_count) desc) Immigrants_rank"
                     "  from f_i94_immigrations f_i94"
                     "  join d_date_arrivals da on da.d_da_id = f_i94.d_da_id  "
                     " group by Year_Month "
                     " order by Year_Month  "
                     )

df_pq3_1.show(20, False)

df_pq3_11 = spark.sql("select da.d_da_year_month as Year_Month"
                     "      ,count(f_i94.f_i94_count) as  Immigrants"
                     "      ,RANK() OVER (ORDER BY count(f_i94.f_i94_count) desc) Immigrants_rank"
                     "  from f_i94_immigrations f_i94"
                     "  join d_date_arrivals da on da.d_da_id = f_i94.d_da_id  "
                     " group by Year_Month "
                     " order by Immigrants_rank  "
                     )

df_pq3_11.show(20, False)<jupyter_output>+----------+----------+---------------+
|Year_Month|Immigrants|Immigrants_rank|
+----------+----------+---------------+
|2016/01   |865969    |11             |
|2016/02   |696579    |12             |
|2016/03   |939031    |9              |
|2016/04   |1042752   |6              |
|2016/05   |1042120   |7              |
|2016/06   |1080648   |4              |
|2016/07   |1328974   |1              |
|2016/08   |1160455   |2              |
|2016/09   |1125245   |3              |
|2016/10   |1050757   |5              |
|2016/11   |904804    |10             |
|2016/12   |991505    |8              |
+----------+----------+---------------+

+----------+----------+---------------+
|Year_Month|Immigrants|Immigrants_rank|
+----------+----------+---------------+
|2016/07   |1328974   |1              |
|2016/08   |1160455   |2              |
|2016/09   |1125245   |3              |
|2016/06   |1080648   |4              |
|2016/10   |1050757   |5              |
|2016/04   |1042752   |6              |[...]<jupyter_text>9. Answer Project Question 3.2: When a foreign person comes to the U.S. for immigration, do they travel on to   another state?<jupyter_code># SQL to answer Project Question 3.2: When a foreign person comes to the U.S. for immigration, do they travel on to
# another state?
df_pq3_2 = spark.sql("select da.d_da_year_month as Year_Month_arrival"
                     "      ,dd.d_dd_year_month as Year_Month_dearture"
                     "      ,count(f_i94.f_i94_count) as Immigrants "
                     "  from f_i94_immigrations f_i94"
                     "  join d_date_arrivals da on da.d_da_id = f_i94.d_da_id  "
                     " left join d_date_departures dd on dd.d_dd_id = f_i94.d_dd_id  "
                     " group by Year_Month_arrival, Year_Month_dearture"
                     " order by Year_Month_arrival, Year_Month_dearture, Immigrants"
                     )

df_pq3_2.show(5000, False)
df_pq3_2 = spark.sql("select da.d_da_year_month as Year_Month_arrival"
                     "      ,dd.d_dd_year_month as Year_Month_dearture"
                     "      ,count(f_i94.f_i94_count) as Immigrants "
                     "  from f_i94_immigrations f_i94"
                     "  join d_date_arrivals da on da.d_da_id = f_i94.d_da_id  "
                     " left join d_date_departures dd on dd.d_dd_id = f_i94.d_dd_id  "
                     " group by Year_Month_arrival, Year_Month_dearture"
                     " order by Immigrants desc "
                     )

df_pq3_2.show(5000, False)<jupyter_output>+------------------+-------------------+----------+
|Year_Month_arrival|Year_Month_dearture|Immigrants|
+------------------+-------------------+----------+
|2016/07           |2016/07            |565670    |
|2016/08           |2016/08            |559430    |
|2016/07           |2016/08            |532022    |
|2016/10           |2016/10            |531023    |
|2016/05           |2016/05            |524489    |
|2016/04           |2016/04            |521618    |
|2016/09           |2016/09            |517067    |
|2016/06           |2016/06            |471452    |
|2016/03           |2016/03            |442997    |
|2016/11           |2016/11            |432226    |
|2016/01           |2016/01            |387914    |
|2016/12           |2017/01            |385122    |
|2016/12           |2016/12            |364770    |
|2016/02           |2016/02            |360331    |
|2016/06           |2016/07            |359382    |
|2016/08           |2016/09            |327683    |
|2016/04    [...]<jupyter_text>10. Answer Project Question 3.3: If a foreign person travels to another state after immigration. After which period of    time does this happen?<jupyter_code># DF to answer Project Question 3.3: If a foreign person travels to another state, after which period of time does this
# happen?
from pyspark.sql.window import Window
from pyspark.sql.functions import dense_rank
windowSpec  = Window.orderBy(col("immigrants").desc())

df_f_i94_immigrations \
    .join(df_d_date_arrivals , df_f_i94_immigrations.d_da_id == df_d_date_arrivals.d_da_id) \
    .join(df_d_date_departures, df_f_i94_immigrations.d_dd_id == df_d_date_departures.d_dd_id) \
    .filter("f_i94_depdate_iso != '1900-01-01'") \
    .withColumn("departure_days_after_arrival", F.datediff(col("f_i94_depdate_iso"), col("f_i94_arrdate_iso"))) \
    .select( "d_da_date"
            ,"d_dd_date"
            ,"departure_days_after_arrival") \
    .groupBy("departure_days_after_arrival").count() \
    .withColumnRenamed("count", "immigrants") \
    .withColumn("dense_rank",dense_rank().over(windowSpec)) \
    .show(500)<jupyter_output>+----------------------------+----------+----------+
|departure_days_after_arrival|immigrants|dense_rank|
+----------------------------+----------+----------+
|                           4|    608273|         1|
|                           5|    574832|         2|
|                           3|    562640|         3|
|                           7|    554430|         4|
|                           6|    533309|         5|
|                           1|    512566|         6|
|                           8|    443648|         7|
|                           2|    417354|         8|
|                           9|    402639|         9|
|                          10|    386824|        10|
|                          14|    372211|        11|
|                          11|    334037|        12|
|                          13|    306093|        13|
|                          12|    305318|        14|
|                          15|    275895|        15|
|                          16|    231645|     [...]
{'fraction_non_alphanumeric': 0.13285057407968281, 'fraction_numerical': 0.05597621079875547, 'mean_word_length': 3.4422700587084147, 'pattern_counts': {'":': 2, '<': 40, '<?xml version=': 0, '>': 62, 'https://': 5, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '155899', 'license': 'MIT', 'n_tokens_mistral': 14384, 'n_tokens_neox': 13031, 'n_words': 2456, 'path': 'P8_Data_Engineering_Capstone_Project/P8_capstone_project/P8_Capstone_Project_Data_Preparation_Step_3.1.3_and_4.1.3.ipynb', 'repo_name': 'peterwissel/Udacity_ND_Data_Engineer', 'size': '36325'}