Data lake with Apache Spark¶

Data Engineering Capstone Project¶

Project Summary¶

The Organization for Tourism Development (OTD) want to analyze migration flux in USA, in order to find insights to significantly and sustainably develop the tourism in USA.

To support their core idea they have identified a set of analysis/queries they want to run on the raw data available.

The project deals with building a data pipeline, to go from raw data to the data insights on the migration flux.

The raw data are gathered from different sources, saved in files and made available for download.

The project shows the execution and decisional flow, specifically:

• Chapter 1: Scope the Project and Gather Data. In Step 1 we give more details on the scope of the project.
  • Describe the data and how they have been obtained
  • Answer the question “how to achieve the target?”
  • What infrastructure (storage, computation, communication) has been used and why
• Chapter 2: Explore and Assess the Data. A first look at the data.
  • Explore the data
  • Check the data for issues, for example null, NaN, or other inconsistencies
• Chapter 3: Define the Data Model. How to organize the data to satisfy the analytical needs.
  • Why this data model has been chosen
  • How it is implemented
• Chapter 4: Run ELT to Model the Data.
  • Load the data from S3 into the SQL database, if any
  • Perform quality checks on the database
  • Perform example queries
• Chapter 5: Complete Project Write Up
  • Documentation of the project
  • Possible scenario extensions

Index¶

• 1. Scope of the Project
• 1.1 What data
• 1.2 What tools
• 1.3 The I94 immigration data
• 1.3.1 What is an I94?
• 1.3.2 The I94 dataset
• 1.3.3 The SAS date format
• 1.3.4 Loading I94 SAS data
• 1.4 World Temperature Data
• 1.5 Airport Code Table
• 1.6 U.S. City Demographic Data
• 2. Data Exploration
• 2.1 The I94 dataset
• 2.2 I94 SAS data load
• 2.3 Explore I94 data
• 2.4 Cleaning the I94 dataset
• 2.5 Store I94 data as parquet
• 2.6 Airport codes dataset: load, clean, save
• 3. The Data Model
• 3.1 Mapping Out Data Pipelines
• 4. Run Pipelines to Model the Data
• 4.1 Provision the AWS S3 infrastructure
• 4.2 Transfer raw data to S3 bucket
• 4.3 EMR cluster on EC2
• 4.3.1 Provision the EMR cluster
• 4.3.2 Coded fields: I94CIT and I94RES
• 4.3.3 Coded field: I94PORT
• 4.3.4 Data cleaning
• 4.3.5 Save clean data (parquet/json) to S3
• 4.3.6 Loading, cleaning and saving airport codes
• 4.4 Querying data on-the-fly
• 4.5 Querying data using the SQL querying style
• 4.6 Data Quality Checks
• 5. Write up
• Lesson learned
• References

1. Scope of the Project¶

The OTD want to run pre-defined queries on the data, with periodical timing.

They also want to maintain the flexibility to run different queries on the data, using BI tools connected to an SQL-like database.

The core data is the dataset provided by US governative agencies filing request of access in the USA (I94 module).

They also have other lower value data available, that are not part of the core analysis, whose use is unclear, therefore are stored in the data lake for a possible future use.

1.1 What data¶

Following datasets are used in the project:

• I94 immigration data for year 2016. Used for the main analysis
• World Temperature Data
• Airport Code Table
• U.S. City Demographic Data

1.2 What tools¶

Because of the nature of the data and the analysis that must be performed, not time-critical analysis, monthly or weekly batch, the choice fell on a cheaper S3-based data lake with on-demand on-the-fly analytical capability: EMR cluster with Apache Spark, and optionally Apache Airflow for scheduled execution (not implemented here).

The architecture shown below has been implemented.

• Starting from a common storage solution (currently Udacity workspace) where both the OTD and its partners have access, the data is then ingested into an S3 bucket, in raw format
• To ease future operations, the data is immediately processed, validated and cleansed using a Spark cluster and stored into S3 in parquet format. Raw and parquet data formats coesist in the data lake.
• By default, the project doesn”t use costly Redshift cluster, but data are queried in-place on the S3 parquet data.
• The EMR cluster serves the analytical needs of the project. SQL based queries are performed using Spark SQL directly on the S3 parquet data
• A Spark job can be triggered monthly, using the Parquet data. The data is aggregated to gain insights on the evolution of the migration flows

1.3 The I94 immigration data¶

The data are provided by the US National Tourism and Trade Office. It is a collection of all I94 that have been filed in 2016.

1.3.1 What is an I94?¶

To give some context is useful to explain what an I94 file is.

From the government website:
“The I-94 is the Arrival/Departure Record, in either paper or electronic format, issued by a Customs and Border Protection (CBP) Officer to foreign visitors entering the United States.”

1.3.2 The I94 dataset¶

Each record contains these fields:

• CICID, unique numer of the file
• I94YR, 4 digit year of the application
• I94MON, Numeric month of the application
• I94CIT, city where the applicant is living
• I94RES, state where the applicant is living
• I94PORT, location (port) where the application is issued
• ARRDATE, arrival date in USA in SAS date format
• I94MODE, how did the applicant arrived in the USA
• I94ADDR, US state where the port is
• DEPDATE is the Departure Date from the USA
• I94BIR, age of applicant in years
• I94VISA, what kind of VISA
• COUNT, used for summary statistics, always 1
• DTADFILE, date added to I-94 Files
• VISAPOST, department of State where where Visa was issued
• OCCUP, occupation that will be performed in U.S.
• ENTDEPA, arrival Flag
• ENTDEPD, departure Flag
• ENTDEPU, update Flag
• MATFLAG, match flag
• BIRYEAR, 4 digit year of birth
• DTADDTO, date to which admitted to U.S. (allowed to stay until)
• GENDER, non-immigrant sex
• INSNUM, INS number
• AIRLINE, airline used to arrive in USA
• ADMNUM, admission Number
• FLTNO, flight number of Airline used to arrive in USA
• VISATYPE, class of admission legally admitting the non-immigrant to temporarily stay in USA

More details in the file I94_SAS_Labels_Descriptions.SAS

1.3.3 The SAS date format¶

Represent any date D0 as the number of days between D0 and the 1th January 1960

1.3.4 Loading I94 SAS data¶

The package saurfang:spark-sas7bdat:2.0.0-s_2.11 and the dependency parso-2.0.8 are needed to read SAS data format.

To load them use the config option spark.jars and give the URL of the repositories, as Spark itself wasn’t able to resolve the dependencies.

1.4 World temperature data¶

The dataset is from Kaggle. It can be found here.

The dataset contains temperature data:

• Global Land and Ocean-and-Land Temperatures (GlobalTemperatures.csv)
• Global Average Land Temperature by Country (GlobalLandTemperaturesByCountry.csv)
• Global Average Land Temperature by State (GlobalLandTemperaturesByState.csv)
• Global Land Temperatures By Major City (GlobalLandTemperaturesByMajorCity.csv)
• Global Land Temperatures By City (GlobalLandTemperaturesByCity.csv)

Below is a snapshop of the global land temperatures by city file.

1.5 Airport codes data¶

This is a table of airport codes, and information on the corresponding cities, like gps coordinates, elevation, country, etc. It comes from Datahub website.

Below is a snapshot of the data.

1.6 U.S. City Demographic Data¶

The dataset comes from OpenSoft. It can be found here.

It contains demographic info on US cities. The info are organized like in the picture below.

2. Data Exploration¶

In this chapter we proceed identifying data quality issues, like missing values, duplicate data, etc.

The purpose is to identify the flow in the data pipeline to programmatically correct data issues.

In this step we work on local data.

2.1 The I94 dataset¶

• How many files are in the I94 dataset?

• What is the size of the files?

2.2 I94 SAS data load¶

To read SAS data format I need to specify the com.github.saurfang.sas.spark format.

• Let’s see the schema Spark applied on reading the file

The most columns are categorical data, this means the information is coded, for example I94CIT=101, 101 is the country code for Albania.

Other columns represent integer data.

It appears clear that there is no need to have data that are defined as double => let’s change those fields to integer

Verifying the schema is correct.

• convert string columns dtadfile and dtaddto to date type

These fields come in a simple string format. To be able to run time-based queries they are converted to date type

• convert columns arrdate and depdate from SAS-date format to a timestamp type.

A date in SAS format is simply the number of days between the chosen date and the reference date (01-01-1960)

• print final schema

2.3 Explore I94 data¶

• How many rows does the I94 database has?

• Let’s see the gender distribution of the applicants

• Where are the I94 applicants coming from?

I want to know the 10 most represented nations

The i94res code 135, where the highest number of visitors come from, corresponds to the the United Kingdom, as can be read in the accompanying file I94_SAS_Labels_Descriptions.SAS

• What port registered the highest number of arrivals?

New York City port registered the highest number of arrivals.

2.4 Cleaning the I94 dataset¶

These are the steps to perform on the I94 database:

1. Identify null and NaN values. Remove duplicates (quality check).
2. Find errors in the records (quality check) for example dates not in year 2016

• Counting how many NaN in each column, excluding the date type columns dtadfile, dtadddto, arrdate, depdate because the isnan function works only on numerical types

• How many rows of the I94 database have null value?

The number of nulls equal the number of rows. It means there is at least one null on each row of the dataframe.

• Now we can count how many null there are in each row

There are many nulls in many columns.

The question is, if there is a need to correct/fill those nulls.

Looking at the data, it seems like some field have been left empty for lack of information.

Because these are categorical data there is no use, at this step, in assigning arbitrary values to the nulls.

The nulls are not going to be filled apriori, but only if a specific need comes up.

• Are there duplicated rows?

Dropping duplicate row

Cheching if the number changed

No row has been dropped => no duplicated row

• Verify that all rows have i94yr column equal 2016

This gives confidence on the consistence of the data

2.5 Store I94 data as parquet¶

I94 data are stored in parquet format in an S3 bucket, they are partinioned using the fields: year, month

2.6 The Airport codes dataset¶

A snippet of the data

How many records?

There are no duplicates

We discover there are some null fields:

The nulls are in these colomuns:

No action taken to fill the nulls

Finally, let’s save the data in parquet format in our temporary folder mimicking the S3 bucket.

3. The Data Model¶

The core of the architecture is a data lake, with S3 storage and EMR processing.

The data are stored into S3 in raw and parquet format.

Apache Spark is the tool elected for analytical tasks, therefore all data are loaded into Spark dataframe using a schema-on-read approach.

For SQL queries style on the data, Spark temporary views are generated.

3.1 Mapping Out Data Pipelines¶

1. Provision the AWS S3 infrastructure
2. Transfer data from the common storage to the S3 lake storage
3. Provision an EMR cluster. It runs 2 steps then autoterminate, these are the 2 steps:
   3.1 Run a spark job to extract codes from file I94_SAS_Labels_Descriptions.SAS and save to S3
   3.2 Data cleaning. Find nan, null, duplicate. Save the clean data to parquet files
4. Generate reports using Spark query on S3 parquet data
5. On-the-fly queries with Spark SQL

4. Run Pipeline to Model the Data¶

4.1 Provision the AWS S3 infrastructure¶

Reading credentials and configuration from file

Create the bucket if it’s not existing

4.2 Transfer raw data to S3 bucket¶

Transfer the data from current shared storage (currently Udacity workspace) to S3 lake storage.

A naive metadata system is implemented. It uses a json file to store basic information on each file added to the S3 bucket:

• file name: file being processed
• added by: user logged as | aws access id
• date added: timestamp of date of processing
• modified on: timestamp of modification time
• notes: any additional information
• access granted to (role or policy): admin | anyone | I94 access policy | weather data access policy |
• expire date: 5 years (default)

These dataset are moved to the S3 lake storage:

• I94 immigration data
• airport codes
• US cities demographics

4.3 EMR cluster on EC2¶

An EMR cluster on EC2 instances with Apache Spark preinstalled is used to perform the ELT work.

A 3-nodes cluster of m5.xlarge istances is configured by default in the config.cfg file.

If the performance requires it, the cluster can be scaled up to use more nodes and/or bigger instances.

After the cluster has been created, the steps to execute spark cleaning jobs are added to the EMR job flow, the steps are in separate .py files. These steps are added:

• Spark job 1
  • extract I94res, i94cit, i94port codes
  • save the codes in a json file in S3
• Spark job 2
  • load I94 raw data from S3
  • change schema
  • data cleaning
  • save parquet data to S3

The cluster is set to auto-terminate by default after executing all the steps.

4.3.1 Provision the EMR cluster¶

Create the cluster using the code emr_cluster.py [Ref. 3] and emr_cluster_spark_submit.py and and set the steps to execute spark_script_1 and spark_script_2.

These scripts have already been previously uploaded to a dedicated folder in the project’s S3 bucket, and are accessible from the EMR cluster.

The file spark_4_emr_codes_extraction.py contains the code for following paragraphs 4.3.1

The file spark_4_emr_I94_processing.py contains the code for following paragraphs 4.3.2, 4.3.3, 4.3.4

4.3.2 Coded fields: I94CIT and I94RES¶

I94CIT, I94RES contain codes indicating the country where the applicant is born (I94CIT), or resident (I94RES).

The data is extracted from I94_SAS_Labels_Descriptions.SAS. This can be done sporadically or every time a change occurred, for example a new code has been added.

The conceptual flow below was implemented.

First steps are define credential to access S3a then load the data in a dataframe, in a single row

Find the section of the file where I94CIT and I94RES are specified.

It start with I94CIT & I94RES and finish with the semicolon character.

To match the section, it is important to have the complete text in a single row, I did this using the option wholetext=True in the previous dataFrame read operation

Now I can split in a dataframe with multiple rows

I filter the rows with structure \ = \

And then create 2 differents columns with code and country

I can finally store the data in a single file in json format

4.3.3 Coded field: I94PORT¶

Similarly to extract the I94PORT codes

The complete code for codes extraction is in spark_4_emr_codes_extraction.py

4.3.4 Data cleaning¶

The cleaning steps have already been shown in section 2, here are only summarized

• Load dataset
• Set schema
  • Numeric fields: double to integer
  • Fields dtadfile and dtaddto: string to date
  • Fields arrdate and depdate: sas to date
• Handle nulls: no fill is set by default
• Drop duplicate

4.3.5 Save clean data (parquet/json) to S3¶

The complete code, refactorized and modularized, is in **spark_4_emr_I94_processing.py**

As a side note, saving the test file as parquet takes about 3 minute on the provisioned cluster. The complete script execution takes 6 minutes.

4.3.6 Loading, cleaning and saving airport codes¶

4.4 Querying data on-the-fly¶

The data in the data lake can be queried on-place. That is the Spark cluster on EMR is directly operating on S3 data.

There are two possible ways to query the data:

• using Spark dataframe functions
• using SQL on tables

We see example of both programming styles.

These are some typical queries that are run on the data:

1. For each port, in a given period, how many arrivals there are in each day?
2. Where are the I94 applicants coming from, in a given period?
3. In the given period, what port registered the highest number of arrivals?
4. Number of arrivals in a given city for a given period
5. Travelers genders
6. Is there a city where the difference between male and female travelers is higher?
7. Find most visited city (the function)

The queries are collected in the Jupyter notebook Capstone project 1 – Querying the data lake.ipynb

4.5 Querying data using the SQL querying style¶

4.6 Data Quality Checks¶

The query-in-place concept implemented here uses a very short pipeline, data are loaded from S3 and after a cleaning process are saved as parquet. Quality of the data is guaranteed by design.

5. Write Up¶

The project has been set up with scalability in mind. All components used, S3 and EMR, offer higher degree of scalability, either horizontal and vertical.

The tool used for the processing, Apache Spark, is the de facto tool for big data processing.

To achieve such a level of scalability we sacrified processing speed. A data warehouse solution with a Redshift database or an OLAP cube would have been faster answering the queries. Anyway nothing forbids to add a DWH to stage the data in case of a more intensive, real-time responsive, usage of the data.

An important part of an ELT/ETL process is automation. Although it has not been touched here, I believe the code developed here is prone to be automatized with a reasonable small effort. A tool like Apache Airflow can be used for the purpose.

Scenario extension¶

• The data was increased by 100x.

In an increased data scenario, the EMR hardware needs to be scaled up accordingly. This is done by simply changing configuration in the config.cfg file. Apache Spark is the tool for big data processing, and is already used as the project analityc tool.

• The data populates a dashboard that must be updated on a daily basis by 7am every day.

In this case an orchestration tool like Apache Airflow is required. A DAG that trigger Phython scripts and Spark jobs executions, needs to be scheduled for daily execution at 7am.

The results of the queries for the dashboard can be saved in a file.

• The database needed to be accessed by 100+ people.

A proper database wasn’t used, on the contrary Amazon S3 is used to store data and queries them in-place. S3 is designed to massive scale in mind, it is able to handle sudden traffic spikes. Therefore, accessing the data by many people shouldn’t be an issue.

The programming used in the project, provision an EMR cluster for any user that plan to run it’s queries. 100+ EMRs is probably going to be expensive for the company. A more efficient sharing of processing resources must be realized.

6. Lessons learned¶

emr 5.28.1 use Python 2 as default¶

• As a consequence important Python packages like pandas are not installed by default for Python 3.
• To install them add an action to the EMR bootstrapping that install all packets
  • install packages for Python 3: python 3 -m pip install \

Adding jars packages to Spark¶

For some reason adding the packages in the Python programm when instantiating the sparkSession doesn’t work (error message package not found).
This doesn’t work:

spark = SparkSession.builder\
                    .config("spark.jars","https://repo1.maven.org/maven2/com/epam/parso/2.0.8/parso-2.0.8.jar,https://repos.spark-packages.org/saurfang/spark-sas7bdat/2.0.0-s_2.11/spark-sas7bdat-2.0.0-s_2.11.jar")\
                    .config("spark.jars.packages","org.apache.hadoop:hadoop-aws:2.7.0")\
                    .enableHiveSupport()\
                    .getOrCreate()

The packages must be added in the spark-submit:

spark-submit -jars https://repo1.maven.org/maven2/com/epam/parso/2.0.8/parso-2.0.8.jar,https://repos.spark-packages.org/saurfang/spark-sas7bdat/2.0.0-s_2.11/spark-sas7bdat-2.0.0-s_2.11.jar s3://helptheplanet/code/spark_4_emr_I94_processing.py

Debugging Spark on EMR¶

While evrything work locally, it doesn’t necessarily means that is going to work on the EMR cluster. Debugging the code is easier with SSH on EMR.

Reading an S3 file from Python is tricky¶

While reading with Spark is straightforward, one just needs to give the address s3://…., with Python boto3 must be used.

Transfering file to S3¶

During the debbuging phase, when the code on S3 must be changed many time using the web interface is slow and unpractical (permanently delete ). Memorize this command:
aws s3 cp <local file> <s3 folder>

Removing the content of a directory from Python¶

import shutil
dirPath = 'metastore_db'
shutil.rmtree(dirPath)

7. References¶

1. AWS CLI Command Reference
2. EMR provisioning is based on: Github repo Boto-3 provisioning
3. Boto3 Command Reference