refactor: restructure monorepo for clean portfolio layout
- Move timesfm-forecast into apps/ directory - Flatten Udacity portfolio projects from deep URL-encoded paths into data-engineering/01-XX numbered directories - Remove old My-Data-Engineering-Portifolio/ parent directory - Rewrite root README.md: professional overview with badges, project table, and repo structure diagram - Create data-engineering/README.md with per-project descriptions - Add README.md for 02-cassandra-modeling (was missing) - Add README.md for 05-airflow-pipelines (was missing) - Normalize capstone readme.md -> README.md - Update .gitignore: add *.cfg, *.env, *.zip, *.sas7bdat, Jupyter checkpoints, IDE dirs; remove uv.lock exclusion - Add dwh.cfg.example and dl.cfg.example credential templates - Untrack real credential files (dwh.cfg, dl.cfg) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@gabriel.pereira
2796
data-engineering/06-immigration-capstone/Capstone Project.ipynb
Normal file
2796
data-engineering/06-immigration-capstone/Capstone Project.ipynb
Normal file
File diff suppressed because one or more lines are too long
File diff suppressed because it is too large
Load Diff
216
data-engineering/06-immigration-capstone/Quality-checks.ipynb
Normal file
216
data-engineering/06-immigration-capstone/Quality-checks.ipynb
Normal file
@@ -0,0 +1,216 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 4,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import configparser\n",
|
||||
"import os\n",
|
||||
"from pathlib import Path\n",
|
||||
"from pyspark.sql import SparkSession"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 9,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# CONFIG\n",
|
||||
"config = configparser.ConfigParser()\n",
|
||||
"config.read('dl.cfg')\n",
|
||||
"\n",
|
||||
"KEY = config.get('AWS', 'AWS_ACCESS_KEY_ID')\n",
|
||||
"SECRET = config.get('AWS', 'AWS_SECRET_ACCESS_KEY')\n",
|
||||
"output_data = './data/outputs'# config.get('S3', 'DEST_S3_BUCKET')\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"os.environ['AWS_ACCESS_KEY_ID']=KEY\n",
|
||||
"os.environ['AWS_SECRET_ACCESS_KEY']=SECRET"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 10,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"spark = SparkSession.builder\\\n",
|
||||
" .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0\")\\\n",
|
||||
" .enableHiveSupport().getOrCreate()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 11,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"s3_bucket = Path(output_data)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 12,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Table: city_code\n",
|
||||
"root\n",
|
||||
" |-- city_code: string (nullable = true)\n",
|
||||
" |-- city: string (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: d_temperature\n",
|
||||
"root\n",
|
||||
" |-- dt: date (nullable = true)\n",
|
||||
" |-- avg_temp: string (nullable = true)\n",
|
||||
" |-- avg_temp_uncertnty: string (nullable = true)\n",
|
||||
" |-- city: string (nullable = true)\n",
|
||||
" |-- country: string (nullable = true)\n",
|
||||
" |-- year: integer (nullable = true)\n",
|
||||
" |-- month: integer (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: d_citizen\n",
|
||||
"root\n",
|
||||
" |-- cic_id: double (nullable = true)\n",
|
||||
" |-- citizen_country: double (nullable = true)\n",
|
||||
" |-- residence_country: double (nullable = true)\n",
|
||||
" |-- birth_year: double (nullable = true)\n",
|
||||
" |-- gender: string (nullable = true)\n",
|
||||
" |-- ins_num: string (nullable = true)\n",
|
||||
" |-- immi_citizen_id: long (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: d_airline\n",
|
||||
"root\n",
|
||||
" |-- cic_id: double (nullable = true)\n",
|
||||
" |-- airline: string (nullable = true)\n",
|
||||
" |-- admin_num: double (nullable = true)\n",
|
||||
" |-- flight_number: string (nullable = true)\n",
|
||||
" |-- visa_type: string (nullable = true)\n",
|
||||
" |-- immi_airline_id: long (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: f_immigration\n",
|
||||
"root\n",
|
||||
" |-- cic_id: double (nullable = true)\n",
|
||||
" |-- year: double (nullable = true)\n",
|
||||
" |-- month: double (nullable = true)\n",
|
||||
" |-- city_code: string (nullable = true)\n",
|
||||
" |-- arrive_date: date (nullable = true)\n",
|
||||
" |-- departure_date: date (nullable = true)\n",
|
||||
" |-- mode: double (nullable = true)\n",
|
||||
" |-- visa: double (nullable = true)\n",
|
||||
" |-- immigration_id: long (nullable = true)\n",
|
||||
" |-- country: string (nullable = true)\n",
|
||||
" |-- state_code: string (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: d_demog_statistics\n",
|
||||
"root\n",
|
||||
" |-- city: string (nullable = true)\n",
|
||||
" |-- state: string (nullable = true)\n",
|
||||
" |-- median_age: string (nullable = true)\n",
|
||||
" |-- avg_household_size: string (nullable = true)\n",
|
||||
" |-- d_demog_statistics: long (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: country_code\n",
|
||||
"root\n",
|
||||
" |-- country_code: string (nullable = true)\n",
|
||||
" |-- country: string (nullable = true)\n",
|
||||
"\n",
|
||||
"Table: state_code\n",
|
||||
"root\n",
|
||||
" |-- state_code: string (nullable = true)\n",
|
||||
" |-- state: string (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"for file_dir in s3_bucket.iterdir():\n",
|
||||
" if file_dir.is_dir():\n",
|
||||
" path = str(file_dir)\n",
|
||||
" df = spark.read.parquet(path)\n",
|
||||
" print(\"Table: \" + path.split('/')[-1])\n",
|
||||
" schema = df.printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 13,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Table: city_code is not empty: total 660 records.\n",
|
||||
"Table: d_temperature is not empty: total 687004 records.\n",
|
||||
"Table: d_citizen is not empty: total 3096313 records.\n",
|
||||
"Table: d_airline is not empty: total 3096313 records.\n",
|
||||
"Table: f_immigration is not empty: total 3096313 records.\n",
|
||||
"Table: d_demog_statistics is not empty: total 596 records.\n",
|
||||
"Table: country_code is not empty: total 235 records.\n",
|
||||
"Table: state_code is not empty: total 55 records.\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"for file_dir in s3_bucket.iterdir():\n",
|
||||
" if file_dir.is_dir():\n",
|
||||
" path = str(file_dir)\n",
|
||||
" df = spark.read.parquet(path)\n",
|
||||
" record_num = df.count()\n",
|
||||
" if record_num <= 0:\n",
|
||||
" raise ValueError(\"This table is empty!\")\n",
|
||||
" else:\n",
|
||||
" print(\"Table: \" + path.split('/')[-1] + f\" is not empty: total {record_num} records.\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": []
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "Python 3",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.6.3"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 4
|
||||
}
|
||||
141
data-engineering/06-immigration-capstone/README.md
Normal file
141
data-engineering/06-immigration-capstone/README.md
Normal file
@@ -0,0 +1,141 @@
|
||||
# Udacity Data Engineer Nanodegree - Capstone Project
|
||||
|
||||
### Project Summary
|
||||
We'll work with four datasets to complete the project. The main dataset will include data on immigration to the United States, and supplementary datasets will include data on airport codes, U.S. city demographics, and temperature data.
|
||||
|
||||
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
|
||||
|
||||
---
|
||||
|
||||
### Step 1: Scope the Project and Gather Data
|
||||
|
||||
#### Scope
|
||||
This project will integrate I94 immigration data, world temperature data and US demographic data to setup a data warehouse with fact and dimension tables.
|
||||
|
||||
* Data Sets
|
||||
1. [I94 Immigration Data](https://travel.trade.gov/research/reports/i94/historical/2016.html)
|
||||
2. [World Temperature Data](https://www.kaggle.com/berkeleyearth/climate-change-earth-surface-temperature-data)
|
||||
3. [U.S. City Demographic Data](https://public.opendatasoft.com/explore/dataset/us-cities-demographics/export/)
|
||||
|
||||
* Tools
|
||||
* AWS S3: data storage
|
||||
* Python for data processing
|
||||
* Pandas - exploratory data analysis on small data set
|
||||
* PySpark - data processing on large data set
|
||||
|
||||
#### Describe and Gather Data
|
||||
|
||||
| Data Set | Format | Description |
|
||||
| --- | --- | --- |
|
||||
|[I94 Immigration Data](https://travel.trade.gov/research/reports/i94/historical/2016.html)| SAS | Data 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).|
|
||||
|[World Temperature Data](https://www.kaggle.com/berkeleyearth/climate-change-earth-surface-temperature-data)| CSV | This dataset is from Kaggle and contains monthly average temperature data at different country in the world wide.|
|
||||
|[U.S. City Demographic Data](https://public.opendatasoft.com/explore/dataset/us-cities-demographics/export/)| CSV | This dataset contains information about the demographics of all US cities and census-designated places with a population greater or equal to 65,000.|
|
||||
|
||||
|
||||
---
|
||||
|
||||
### Step 2: Explore and Assess the Data
|
||||
|
||||
#### Explore the Data
|
||||
|
||||
1. Use pandas for exploratory data analysis to get an overview on these data sets
|
||||
2. Split data sets to dimensional tables and change column names for better understanding
|
||||
3. Utilize PySpark on one of the SAS data sets to test ETL data pipeline logic
|
||||
|
||||
#### Cleaning Steps
|
||||
|
||||
1. Transform arrdate, depdate from SAS time format to pandad.datetime
|
||||
2. Parse description file to get auxiliary dimension table - country_code, city_code, state_code, mode, visa
|
||||
3. Tranform city, state to upper case to match city _code and state _code table
|
||||
|
||||
Please refer to [Capstone_Project.ipynb](./Capstone_Project.ipynb).
|
||||
|
||||
(This step was completed in Udacity workspace as pre-steps for building up and testing the ETL data pipeline. File paths should be modified if notebook is run locally.)
|
||||
|
||||
---
|
||||
|
||||
### Step 3: Define the Data Model
|
||||
|
||||
#### Conceptual Data Model
|
||||
Since the purpose of this data lake is for BI app usage, we will model these data sets with star schema data modeling.
|
||||
|
||||
* Star Schema
|
||||
|
||||
|
||||
|
||||
#### Data Pipeline Build Up Steps
|
||||
|
||||
1. Assume all data sets are stored in S3 buckets
|
||||
2. Follow by Step 2 – Cleaning step to clean up data sets
|
||||
3. Transform immigration data to 1 fact table and 2 dimension tables, fact table will be partitioned by `state_code`
|
||||
4. Parsing label description file to get auxiliary tables
|
||||
5. Transform temperature data to dimension table
|
||||
6. Split demography data to 2 dimension tables
|
||||
7. Store these tables back to target S3 bucket
|
||||
|
||||
---
|
||||
|
||||
### Step 4: Run Pipelines to Model the Data
|
||||
|
||||
Please refer to [etl.py](./etl.py).
|
||||
|
||||
#### 4.1 Create the data model
|
||||
|
||||
Data processing and data model was created by Spark.
|
||||
|
||||
Please refer to [Capstone_Project.ipynb](./Capstone_Project.ipynb).
|
||||
|
||||
#### 4.2 Data Quality Checks
|
||||
|
||||
Data quality checks includes
|
||||
|
||||
1. No empty table after running ETL data pipeline
|
||||
2. Data schema of every dimensional table matches data model
|
||||
|
||||
Please refer to [Quality-checks.ipynb](./Quality-checks.ipynb).
|
||||
|
||||
#### 4.3 Data dictionary
|
||||
|
||||
Please refer to [Capstone_Project.ipynb](./Capstone_Project.ipynb).
|
||||
|
||||
---
|
||||
|
||||
### Step 5: Complete Project Write Up
|
||||
|
||||
#### Tools and Technologies
|
||||
1. AWS S3 for data storage
|
||||
2. Pandas for sample data set exploratory data analysis
|
||||
3. PySpark for large data set data processing to transform staging table to dimensional table
|
||||
|
||||
#### Data Update Frequency
|
||||
1. Tables created from immigration and temperature data set should be updated monthly since the raw data set is built up monthly.
|
||||
2. Tables created from demography data set could be updated annually since demography data collection takes time and high frequent demography might take high cost but generate wrong conclusion.
|
||||
3. All tables should be update in an append-only mode.
|
||||
|
||||
#### Future Design Considerations
|
||||
1. The data was increased by 100x.
|
||||
|
||||
If Spark with standalone server mode can not process 100x data set, we could consider to put data in [AWS EMR](https://aws.amazon.com/tw/emr/?nc2=h_ql_prod_an_emr&whats-new-cards.sort-by=item.additionalFields.postDateTime&whats-new-cards.sort-order=desc) which is a distributed data cluster for processing large data sets on cloud
|
||||
|
||||
2. The data populates a dashboard that must be updated on a daily basis by 7am every day.
|
||||
|
||||
[Apache Airflow](https://airflow.apache.org) could be used for building up a ETL data pipeline to regularly update the date and populate a report. Apache Airflow also integrate with Python and AWS very well. More applications can be combined together to deliever more powerful task automation.
|
||||
|
||||
3. The database needed to be accessed by 100+ people.
|
||||
|
||||
[AWS Redshift](https://aws.amazon.com/tw/redshift/?nc2=h_ql_prod_db_rs&whats-new-cards.sort-by=item.additionalFields.postDateTime&whats-new-cards.sort-order=desc) can handle up to 500 connections. If this SSOT database will be accessed by 100+ people, we can move this database to Redshift with confidence to handle this request. Cost/Benefit analysis will be needed if we are going be implement this cloud solution.
|
||||
|
||||
---
|
||||
|
||||
### Future Improvements
|
||||
There are several incompletions within these data sets. We will need to collect more data to get accurate.
|
||||
|
||||
1. Immigration data set is based at 2016 but temperature data set only get to 2013 which is not enough for us to see the temperature change at 2016.
|
||||
|
||||
2. Missing state and city in label description file. This makes it hard to join immigration tables and demography tables.
|
||||
55076
data-engineering/06-immigration-capstone/airport-codes_csv.csv
Normal file
55076
data-engineering/06-immigration-capstone/airport-codes_csv.csv
Normal file
File diff suppressed because it is too large
Load Diff
BIN
data-engineering/06-immigration-capstone/data-model.png
Normal file
BIN
data-engineering/06-immigration-capstone/data-model.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 169 KiB |
10
data-engineering/06-immigration-capstone/dl.cfg.example
Normal file
10
data-engineering/06-immigration-capstone/dl.cfg.example
Normal file
@@ -0,0 +1,10 @@
|
||||
[AWS]
|
||||
AWS_ACCESS_KEY_ID=YOUR_AWS_ACCESS_KEY_ID
|
||||
AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET_ACCESS_KEY
|
||||
|
||||
[IAM_ROLE]
|
||||
ARN=arn:aws:iam::YOUR_ACCOUNT_ID:role/myRedshiftRole
|
||||
|
||||
[S3]
|
||||
SOURCE_S3_BUCKET=s3a://your-bucket/
|
||||
DEST_S3_BUCKET=s3a://your-bucket/
|
||||
194
data-engineering/06-immigration-capstone/etl-s3-testing.ipynb
Normal file
194
data-engineering/06-immigration-capstone/etl-s3-testing.ipynb
Normal file
@@ -0,0 +1,194 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"source": [
|
||||
"### ETL notebook for testing the pipeline"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 1,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import configparser\n",
|
||||
"import pandas as pd\n",
|
||||
"from datetime import datetime\n",
|
||||
"from pyspark.sql.functions import dayofweek\n",
|
||||
"import os\n",
|
||||
"from pyspark.sql import SparkSession\n",
|
||||
"from pyspark.sql.functions import udf, col, lit\n",
|
||||
"from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format, to_date, upper\n",
|
||||
"import logging\n",
|
||||
"from pyspark.sql.types import DateType\n",
|
||||
"from pyspark.sql.functions import monotonically_increasing_id"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 2,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# CONFIG\n",
|
||||
"config = configparser.ConfigParser()\n",
|
||||
"config.read('dl.cfg')\n",
|
||||
"\n",
|
||||
"KEY = config.get('AWS', 'AWS_ACCESS_KEY_ID')\n",
|
||||
"SECRET = config.get('AWS', 'AWS_SECRET_ACCESS_KEY')\n",
|
||||
"output_data = config.get('S3', 'DEST_S3_BUCKET')\n",
|
||||
"\n",
|
||||
"\n",
|
||||
"os.environ['AWS_ACCESS_KEY_ID']=KEY\n",
|
||||
"os.environ['AWS_SECRET_ACCESS_KEY']=SECRET\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 3,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def create_spark_session():\n",
|
||||
" spark = SparkSession \\\n",
|
||||
" .builder \\\n",
|
||||
" .config(\"spark.jars.repositories\", \"https://repos.spark-packages.org/\")\\\n",
|
||||
" .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0,saurfang:spark-sas7bdat:2.0.0-s_2.11\")\\\n",
|
||||
" .enableHiveSupport().getOrCreate()\n",
|
||||
" return spark"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 4,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"spark = create_spark_session()\n",
|
||||
"input_data ='./'\n",
|
||||
"output_data = 's3a://gfp-udacity/testing'"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 5,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def rename_columns(table, new_columns):\n",
|
||||
" for original, new in zip(table.columns, new_columns):\n",
|
||||
" table = table.withColumnRenamed(original, new)\n",
|
||||
" return table"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 6,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def process_demography_data(spark, input_data, output_data):\n",
|
||||
" \"\"\" Process demograpy data to get dim_demog_population \n",
|
||||
" and d_demog_statistics table\n",
|
||||
" Arguments:\n",
|
||||
" spark {object}: SparkSession object\n",
|
||||
" input_data {object}: Source S3 endpoint\n",
|
||||
" output_data {object}: Target S3 endpoint\n",
|
||||
" Returns:\n",
|
||||
" None\n",
|
||||
" \"\"\"\n",
|
||||
"\n",
|
||||
" logging.info(\"Start processing d_demog_statistics\")\n",
|
||||
" # read demography data file\n",
|
||||
" demog_data = os.path.join(input_data + 'us-cities-demographics.csv')\n",
|
||||
" df = spark.read.format('csv').options(header=True, delimiter=';').load(demog_data)\n",
|
||||
"\n",
|
||||
"\n",
|
||||
" d_demog_statistics = df.select(['City', 'State', 'Male Population', 'Female Population', \\\n",
|
||||
" 'Number of Veterans', 'Foreign-born', 'Race']).distinct() \\\n",
|
||||
" .withColumn(\"demog_pop_id\", monotonically_increasing_id())\n",
|
||||
"\n",
|
||||
"\n",
|
||||
" new_columns = ['city', 'state', 'male_population', 'female_population', \\\n",
|
||||
" 'num_vetarans', 'foreign_born', 'race']\n",
|
||||
" d_demog_statistics = rename_columns(d_demog_statistics, new_columns)\n",
|
||||
"\n",
|
||||
" # write dim_demog_population table to parquet files\n",
|
||||
" d_demog_statistics.write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'd_demog_statistics')\n",
|
||||
"\n",
|
||||
" \n",
|
||||
" logging.info(\"Start processing d_demog_statistics\")\n",
|
||||
" d_demog_statistics = df.select(['City', 'State', 'Median Age', 'Average Household Size'])\\\n",
|
||||
" .distinct()\\\n",
|
||||
" .withColumn(\"d_demog_statistics\", monotonically_increasing_id())\n",
|
||||
"\n",
|
||||
" new_columns = ['city', 'state', 'median_age', 'avg_household_size']\n",
|
||||
" d_demog_statistics = rename_columns(d_demog_statistics, new_columns)\n",
|
||||
" d_demog_statistics = d_demog_statistics.withColumn('city', upper(col('city')))\n",
|
||||
" d_demog_statistics = d_demog_statistics.withColumn('state', upper(col('state')))\n",
|
||||
"\n",
|
||||
" # write dim_demog_statistics table to parquet files\n",
|
||||
" d_demog_statistics.write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 7,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"process_demography_data(spark, input_data, output_data)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": []
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "Python 3",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.6.3"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 4
|
||||
}
|
||||
572
data-engineering/06-immigration-capstone/etl.ipynb
Normal file
572
data-engineering/06-immigration-capstone/etl.ipynb
Normal file
@@ -0,0 +1,572 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"source": [
|
||||
"### ETL notebook for testing the pipeline"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 1,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import configparser\n",
|
||||
"import pandas as pd\n",
|
||||
"from datetime import datetime\n",
|
||||
"from pyspark.sql.functions import dayofweek\n",
|
||||
"import os\n",
|
||||
"from pyspark.sql import SparkSession\n",
|
||||
"from pyspark.sql.functions import udf, col, lit\n",
|
||||
"from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format, to_date, upper\n",
|
||||
"import logging\n",
|
||||
"from pyspark.sql.types import DateType\n",
|
||||
"from pyspark.sql.functions import monotonically_increasing_id"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 2,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def create_spark_session():\n",
|
||||
" spark = SparkSession \\\n",
|
||||
" .builder \\\n",
|
||||
" .config(\"spark.jars.repositories\", \"https://repos.spark-packages.org/\")\\\n",
|
||||
" .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0,saurfang:spark-sas7bdat:2.0.0-s_2.11\")\\\n",
|
||||
" .enableHiveSupport().getOrCreate()\n",
|
||||
" return spark"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 3,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def rename_columns(table, new_columns):\n",
|
||||
" for original, new in zip(table.columns, new_columns):\n",
|
||||
" table = table.withColumnRenamed(original, new)\n",
|
||||
" return table"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 4,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def SAS_to_date(date):\n",
|
||||
" if date is not None:\n",
|
||||
" return pd.to_timedelta(date, unit='D') + pd.Timestamp('1960-1-1')\n",
|
||||
"\n",
|
||||
"SAS_to_date_udf = udf(SAS_to_date, DateType())"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 5,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def process_immigration_data(spark, output_data):\n",
|
||||
" \"\"\"Process immigration data to get f_immigration, d_immi_citzen and d_immi_airline tables\n",
|
||||
" Arguments:\n",
|
||||
" spark {object}: SparkSession object\n",
|
||||
" input_data {object}: Source S3 endpoint\n",
|
||||
" output_data {object}: Target S3 endpoint\n",
|
||||
" Returns:\n",
|
||||
" None\n",
|
||||
" \"\"\"\n",
|
||||
" logging.info(\"Start processing immigration\")\n",
|
||||
" \n",
|
||||
" # read immigration data file\n",
|
||||
" df = spark.read.format(\"com.github.saurfang.sas.spark\").load(\"../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat\", forceLowercaseNames=True, inferLong=True)\n",
|
||||
" \n",
|
||||
" logging.info(\"Start processing f_immigration\")\n",
|
||||
" \n",
|
||||
" # extract columns to create fact_immigration table\n",
|
||||
" f_immigration = df.select('cicid', 'i94yr', 'i94mon', 'i94port', 'i94addr', 'arrdate', 'depdate', 'i94mode', 'i94visa')\n",
|
||||
" f_immigration = f_immigration.distinct()\n",
|
||||
" f_immigration = f_immigration.withColumn(\"immigration_id\", monotonically_increasing_id())\n",
|
||||
" \n",
|
||||
" # data wrangling to match data model\n",
|
||||
" new_columns = ['cic_id', 'year', 'month', 'city_code', 'state_code', 'arrive_date', 'departure_date', 'mode', 'visa']\n",
|
||||
" \n",
|
||||
" # renaming columns using the function rename_columns()\n",
|
||||
" f_immigration = rename_columns(f_immigration, new_columns)\n",
|
||||
" \n",
|
||||
" # add a new column to f_immigration by assigning a literal or constant value = United States\n",
|
||||
" f_immigration = f_immigration.withColumn('country', lit('United States'))\n",
|
||||
" \n",
|
||||
" # convert column arrive_date to date type format\n",
|
||||
" f_immigration = f_immigration.withColumn('arrive_date', SAS_to_date_udf(col('arrive_date')))\n",
|
||||
" \n",
|
||||
" # convert column departure_date to date type format\n",
|
||||
" f_immigration = f_immigration.withColumn('departure_date', SAS_to_date_udf(col('departure_date')))\n",
|
||||
" \n",
|
||||
" logging.info(\"Start loading f_immigration parquet files partitioned by state_code\")\n",
|
||||
" \n",
|
||||
" # write f_immigration table to parquet files partitioned by state_code\n",
|
||||
" f_immigration.write.mode(\"overwrite\").partitionBy('state_code').parquet(path=output_data + 'f_immigration') \n",
|
||||
" \n",
|
||||
" \n",
|
||||
" \n",
|
||||
" logging.info(\"Start processing d_citizen table\")\n",
|
||||
" \n",
|
||||
" # extract columns from immigration data file to create d_citizen table\n",
|
||||
" d_citizen = df.select('cicid', 'i94cit', 'i94res', 'biryear', 'gender', 'insnum').distinct().withColumn(\"immi_citizen_id\", monotonically_increasing_id())\n",
|
||||
" \n",
|
||||
" # data wrangling to match data model\n",
|
||||
" new_columns = ['cic_id', 'citizen_country', 'residence_country', 'birth_year', 'gender', 'ins_num']\n",
|
||||
" d_citizen = rename_columns(d_citizen, new_columns)\n",
|
||||
"\n",
|
||||
" # write d_citizen table to parquet files\n",
|
||||
" d_citizen.write.mode(\"overwrite\").parquet(path=output_data + 'd_citizen')\n",
|
||||
" \n",
|
||||
" \n",
|
||||
" \n",
|
||||
" logging.info(\"Start processing d_airline\")\n",
|
||||
" \n",
|
||||
" # extract columns from immigration data file to create d_airline table\n",
|
||||
" d_airline = df.select('cicid', 'airline', 'admnum', 'fltno', 'visatype').distinct().withColumn(\"immi_airline_id\", monotonically_increasing_id())\n",
|
||||
" \n",
|
||||
" # data wrangling to match data model\n",
|
||||
" new_columns = ['cic_id', 'airline', 'admin_num', 'flight_number', 'visa_type']\n",
|
||||
" d_airline = rename_columns(d_airline, new_columns)\n",
|
||||
"\n",
|
||||
" # write d_airline table to parquet files\n",
|
||||
" d_airline.write.mode(\"overwrite\").parquet(path=output_data + 'd_airline')"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 6,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def process_label_descriptions(spark, input_data, output_data):\n",
|
||||
" \"\"\" Parsing label desctiption file to get codes of country, city, state\n",
|
||||
" Arguments:\n",
|
||||
" spark {object}: SparkSession object\n",
|
||||
" input_data {object}: Source S3 endpoint\n",
|
||||
" output_data {object}: Target S3 endpoint\n",
|
||||
" Returns:\n",
|
||||
" None\n",
|
||||
" \"\"\"\n",
|
||||
"\n",
|
||||
" logging.info(\"Start processing label descriptions\")\n",
|
||||
" label_file = os.path.join(input_data + \"I94_SAS_Labels_Descriptions.SAS\")\n",
|
||||
" with open(label_file) as f:\n",
|
||||
" contents = f.readlines()\n",
|
||||
"\n",
|
||||
" country_code = {}\n",
|
||||
" for countries in contents[10:245]:\n",
|
||||
" pair = countries.split('=')\n",
|
||||
" code, country = pair[0].strip(), pair[1].strip().strip(\"'\")\n",
|
||||
" country_code[code] = country\n",
|
||||
" \n",
|
||||
" spark.createDataFrame(country_code.items(), ['country_code', 'country'])\\\n",
|
||||
" .write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'country_code')\n",
|
||||
"\n",
|
||||
" city_code = {}\n",
|
||||
" for cities in contents[302:962]:\n",
|
||||
" pair = cities.split('=')\n",
|
||||
" code, city = pair[0].strip(\"\\t\").strip().strip(\"'\"),\\\n",
|
||||
" pair[1].strip('\\t').strip().strip(\"''\")\n",
|
||||
" city_code[code] = city\n",
|
||||
" spark.createDataFrame(city_code.items(), ['city_code', 'city'])\\\n",
|
||||
" .write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'city_code')\n",
|
||||
"\n",
|
||||
" state_code = {}\n",
|
||||
" for states in contents[981:1036]:\n",
|
||||
" pair = states.split('=')\n",
|
||||
" code, state = pair[0].strip('\\t').strip(\"'\"), pair[1].strip().strip(\"'\")\n",
|
||||
" state_code[code] = state\n",
|
||||
" spark.createDataFrame(state_code.items(), ['state_code', 'state'])\\\n",
|
||||
" .write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'state_code')"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 7,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def process_temperature_data(spark, output_data):\n",
|
||||
" \"\"\" Process temperature data to get dim_temperature table\n",
|
||||
" Arguments:\n",
|
||||
" spark {object}: SparkSession object\n",
|
||||
" input_data {object}: Source S3 endpoint\n",
|
||||
" output_data {object}: Target S3 endpoint\n",
|
||||
" Returns:\n",
|
||||
" None\n",
|
||||
" \"\"\"\n",
|
||||
"\n",
|
||||
" logging.info(\"Start processing d_temperature\")\n",
|
||||
" # read temperature data file\n",
|
||||
" tempe_data = os.path.join('../../data2/GlobalLandTemperaturesByCity.csv')\n",
|
||||
" df = spark.read.csv(tempe_data, header=True)\n",
|
||||
"\n",
|
||||
" df = df.where(df['Country'] == 'United States')\n",
|
||||
" d_temperature = df.select(['dt', 'AverageTemperature', 'AverageTemperatureUncertainty',\\\n",
|
||||
" 'City', 'Country']).distinct()\n",
|
||||
"\n",
|
||||
" new_columns = ['dt', 'avg_temp', 'avg_temp_uncertnty', 'city', 'country']\n",
|
||||
" d_temperature = rename_columns(d_temperature, new_columns)\n",
|
||||
"\n",
|
||||
" d_temperature = d_temperature.withColumn('dt', to_date(col('dt')))\n",
|
||||
" d_temperature = d_temperature.withColumn('year', year(d_temperature['dt']))\n",
|
||||
" d_temperature = d_temperature.withColumn('month', month(d_temperature['dt']))\n",
|
||||
" \n",
|
||||
" # write dim_temperature table to parquet files\n",
|
||||
" d_temperature.write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'd_temperature')"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 8,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def process_demography_data(spark, input_data, output_data):\n",
|
||||
" \"\"\" Process demograpy data to get dim_demog_population \n",
|
||||
" and d_demog_statistics table\n",
|
||||
" Arguments:\n",
|
||||
" spark {object}: SparkSession object\n",
|
||||
" input_data {object}: Source S3 endpoint\n",
|
||||
" output_data {object}: Target S3 endpoint\n",
|
||||
" Returns:\n",
|
||||
" None\n",
|
||||
" \"\"\"\n",
|
||||
"\n",
|
||||
" logging.info(\"Start processing d_demog_statistics\")\n",
|
||||
" # read demography data file\n",
|
||||
" demog_data = os.path.join(input_data + 'us-cities-demographics.csv')\n",
|
||||
" df = spark.read.format('csv').options(header=True, delimiter=';').load(demog_data)\n",
|
||||
"\n",
|
||||
"\n",
|
||||
" d_demog_statistics = df.select(['City', 'State', 'Male Population', 'Female Population', \\\n",
|
||||
" 'Number of Veterans', 'Foreign-born', 'Race']).distinct() \\\n",
|
||||
" .withColumn(\"demog_pop_id\", monotonically_increasing_id())\n",
|
||||
"\n",
|
||||
"\n",
|
||||
" new_columns = ['city', 'state', 'male_population', 'female_population', \\\n",
|
||||
" 'num_vetarans', 'foreign_born', 'race']\n",
|
||||
" d_demog_statistics = rename_columns(d_demog_statistics, new_columns)\n",
|
||||
"\n",
|
||||
" # write dim_demog_population table to parquet files\n",
|
||||
" d_demog_statistics.write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'd_demog_statistics')\n",
|
||||
"\n",
|
||||
" \n",
|
||||
" logging.info(\"Start processing d_demog_statistics\")\n",
|
||||
" d_demog_statistics = df.select(['City', 'State', 'Median Age', 'Average Household Size'])\\\n",
|
||||
" .distinct()\\\n",
|
||||
" .withColumn(\"d_demog_statistics\", monotonically_increasing_id())\n",
|
||||
"\n",
|
||||
" new_columns = ['city', 'state', 'median_age', 'avg_household_size']\n",
|
||||
" d_demog_statistics = rename_columns(d_demog_statistics, new_columns)\n",
|
||||
" d_demog_statistics = d_demog_statistics.withColumn('city', upper(col('city')))\n",
|
||||
" d_demog_statistics = d_demog_statistics.withColumn('state', upper(col('state')))\n",
|
||||
"\n",
|
||||
" # write dim_demog_statistics table to parquet files\n",
|
||||
" d_demog_statistics.write.mode(\"overwrite\")\\\n",
|
||||
" .parquet(path=output_data + 'd_demog_statistics')"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 9,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"'''Paths for local testing'''\n",
|
||||
"input_data = \"./\"\n",
|
||||
"output_data = \"./data/outputs/\"\n",
|
||||
"\n",
|
||||
"spark = create_spark_session()\n",
|
||||
"\n",
|
||||
"process_immigration_data(spark, output_data)\n",
|
||||
"process_label_descriptions(spark, input_data, output_data)\n",
|
||||
"process_temperature_data(spark, output_data)\n",
|
||||
"process_demography_data(spark, input_data, output_data)\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"source": [
|
||||
"### Data Model Schema"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 11,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- cic_id: double (nullable = true)\n",
|
||||
" |-- year: double (nullable = true)\n",
|
||||
" |-- month: double (nullable = true)\n",
|
||||
" |-- city_code: string (nullable = true)\n",
|
||||
" |-- arrive_date: date (nullable = true)\n",
|
||||
" |-- departure_date: date (nullable = true)\n",
|
||||
" |-- mode: double (nullable = true)\n",
|
||||
" |-- visa: double (nullable = true)\n",
|
||||
" |-- immigration_id: long (nullable = true)\n",
|
||||
" |-- country: string (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# f_immigration table sample\n",
|
||||
"spark.read.parquet('data/outputs/f_immigration/state_code=AE/part-00000-79a43b03-8ba1-418b-bd9e-b4b3de3d441c.c000.snappy.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 12,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- cic_id: double (nullable = true)\n",
|
||||
" |-- citizen_country: double (nullable = true)\n",
|
||||
" |-- residence_country: double (nullable = true)\n",
|
||||
" |-- birth_year: double (nullable = true)\n",
|
||||
" |-- gender: string (nullable = true)\n",
|
||||
" |-- ins_num: string (nullable = true)\n",
|
||||
" |-- immi_citizen_id: long (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# d_citizen table sample\n",
|
||||
"spark.read.parquet('data/outputs/d_citizen/part-00000-ddacc44c-bdd0-4694-a9fa-4a367a610df6-c000.snappy.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 13,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- cic_id: double (nullable = true)\n",
|
||||
" |-- airline: string (nullable = true)\n",
|
||||
" |-- admin_num: double (nullable = true)\n",
|
||||
" |-- flight_number: string (nullable = true)\n",
|
||||
" |-- visa_type: string (nullable = true)\n",
|
||||
" |-- immi_airline_id: long (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# d_airline table sample\n",
|
||||
"spark.read.parquet('data/outputs/d_airline/part-00000-5464a16a-e021-494e-9e76-d6df7e80c20f-c000.snappy.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 14,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- country_code: string (nullable = true)\n",
|
||||
" |-- country: string (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# country_code table sample\n",
|
||||
"spark.read.parquet('data/outputs/country_code/*.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 15,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- city_code: string (nullable = true)\n",
|
||||
" |-- city: string (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# city_code table sample\n",
|
||||
"spark.read.parquet('data/outputs/city_code/*.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 16,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- state_code: string (nullable = true)\n",
|
||||
" |-- state: string (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# state_code table sample\n",
|
||||
"spark.read.parquet('data/outputs/state_code/*.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 17,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- dt: date (nullable = true)\n",
|
||||
" |-- avg_temp: string (nullable = true)\n",
|
||||
" |-- avg_temp_uncertnty: string (nullable = true)\n",
|
||||
" |-- city: string (nullable = true)\n",
|
||||
" |-- country: string (nullable = true)\n",
|
||||
" |-- year: integer (nullable = true)\n",
|
||||
" |-- month: integer (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# d_temperature table sample\n",
|
||||
"spark.read.parquet('data/outputs/d_temperature/*.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 18,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"root\n",
|
||||
" |-- city: string (nullable = true)\n",
|
||||
" |-- state: string (nullable = true)\n",
|
||||
" |-- median_age: string (nullable = true)\n",
|
||||
" |-- avg_household_size: string (nullable = true)\n",
|
||||
" |-- d_demog_statistics: long (nullable = true)\n",
|
||||
"\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# d_demog_statistics table sample\n",
|
||||
"spark.read.parquet('data/outputs/d_demog_statistics/*.parquet').printSchema()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {
|
||||
"editable": true
|
||||
},
|
||||
"outputs": [],
|
||||
"source": []
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "Python 3",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.6.3"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 4
|
||||
}
|
||||
248
data-engineering/06-immigration-capstone/etl.py
Normal file
248
data-engineering/06-immigration-capstone/etl.py
Normal file
@@ -0,0 +1,248 @@
|
||||
import configparser
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
from pyspark.sql.functions import dayofweek
|
||||
import os
|
||||
from pyspark.sql import SparkSession
|
||||
from pyspark.sql.functions import udf, col, lit
|
||||
from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format, to_date, upper
|
||||
import logging
|
||||
from pyspark.sql.types import DateType
|
||||
from pyspark.sql.functions import monotonically_increasing_id
|
||||
|
||||
|
||||
# CONFIG
|
||||
config = configparser.ConfigParser()
|
||||
config.read('dl.cfg')
|
||||
|
||||
KEY = config.get('AWS', 'AWS_ACCESS_KEY_ID')
|
||||
SECRET = config.get('AWS', 'AWS_SECRET_ACCESS_KEY')
|
||||
output_data = config.get('S3', 'DEST_S3_BUCKET')
|
||||
|
||||
|
||||
os.environ['AWS_ACCESS_KEY_ID']=KEY
|
||||
os.environ['AWS_SECRET_ACCESS_KEY']=SECRET
|
||||
|
||||
def create_spark_session():
|
||||
spark = SparkSession \
|
||||
.builder \
|
||||
.config("spark.jars.repositories", "https://repos.spark-packages.org/")\
|
||||
.config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:2.7.0,saurfang:spark-sas7bdat:2.0.0-s_2.11")\
|
||||
.enableHiveSupport().getOrCreate()
|
||||
return spark
|
||||
|
||||
|
||||
def rename_columns(table, new_columns):
|
||||
for original, new in zip(table.columns, new_columns):
|
||||
table = table.withColumnRenamed(original, new)
|
||||
return table
|
||||
|
||||
|
||||
def SAS_to_date(date):
|
||||
if date is not None:
|
||||
return pd.to_timedelta(date, unit='D') + pd.Timestamp('1960-1-1')
|
||||
|
||||
SAS_to_date_udf = udf(SAS_to_date, DateType())
|
||||
|
||||
|
||||
def process_immigration_data(spark, output_data):
|
||||
"""Process immigration data to get f_immigration, d_immi_citzen and d_immi_airline tables
|
||||
Arguments:
|
||||
spark {object}: SparkSession object
|
||||
input_data {object}: Source S3 endpoint
|
||||
output_data {object}: Target S3 endpoint
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
logging.info("Start processing immigration")
|
||||
|
||||
# read immigration data file
|
||||
df = spark.read.format("com.github.saurfang.sas.spark").load("../../data/18-83510-I94-Data-2016/i94_apr16_sub.sas7bdat", forceLowercaseNames=True, inferLong=True)
|
||||
|
||||
logging.info("Start processing f_immigration")
|
||||
|
||||
# extract columns to create fact_immigration table
|
||||
f_immigration = df.select('cicid', 'i94yr', 'i94mon', 'i94port', 'i94addr', 'arrdate', 'depdate', 'i94mode', 'i94visa')
|
||||
f_immigration = f_immigration.distinct()
|
||||
f_immigration = f_immigration.withColumn("immigration_id", monotonically_increasing_id())
|
||||
|
||||
# data wrangling to match data model
|
||||
new_columns = ['cic_id', 'year', 'month', 'city_code', 'state_code', 'arrive_date', 'departure_date', 'mode', 'visa']
|
||||
|
||||
# renaming columns using the function rename_columns()
|
||||
f_immigration = rename_columns(f_immigration, new_columns)
|
||||
|
||||
# add a new column to f_immigration by assigning a literal or constant value = United States
|
||||
f_immigration = f_immigration.withColumn('country', lit('United States'))
|
||||
|
||||
# convert column arrive_date to date type format
|
||||
f_immigration = f_immigration.withColumn('arrive_date', SAS_to_date_udf(col('arrive_date')))
|
||||
|
||||
# convert column departure_date to date type format
|
||||
f_immigration = f_immigration.withColumn('departure_date', SAS_to_date_udf(col('departure_date')))
|
||||
|
||||
logging.info("Start loading f_immigration parquet files partitioned by state_code")
|
||||
|
||||
# write f_immigration table to parquet files partitioned by state_code
|
||||
f_immigration.write.mode("overwrite").partitionBy('state_code').parquet(path=output_data + 'f_immigration')
|
||||
|
||||
|
||||
|
||||
logging.info("Start processing d_citizen table")
|
||||
|
||||
# extract columns from immigration data file to create d_citizen table
|
||||
d_citizen = df.select('cicid', 'i94cit', 'i94res', 'biryear', 'gender', 'insnum').distinct().withColumn("immi_citizen_id", monotonically_increasing_id())
|
||||
|
||||
# data wrangling to match data model
|
||||
new_columns = ['cic_id', 'citizen_country', 'residence_country', 'birth_year', 'gender', 'ins_num']
|
||||
d_citizen = rename_columns(d_citizen, new_columns)
|
||||
|
||||
# write d_citizen table to parquet files
|
||||
d_citizen.write.mode("overwrite").parquet(path=output_data + 'd_citizen')
|
||||
|
||||
|
||||
|
||||
logging.info("Start processing d_airline")
|
||||
|
||||
# extract columns from immigration data file to create d_airline table
|
||||
d_airline = df.select('cicid', 'airline', 'admnum', 'fltno', 'visatype').distinct().withColumn("immi_airline_id", monotonically_increasing_id())
|
||||
|
||||
# data wrangling to match data model
|
||||
new_columns = ['cic_id', 'airline', 'admin_num', 'flight_number', 'visa_type']
|
||||
d_airline = rename_columns(d_airline, new_columns)
|
||||
|
||||
# write d_airline table to parquet files
|
||||
d_airline.write.mode("overwrite").parquet(path=output_data + 'd_airline')
|
||||
|
||||
|
||||
|
||||
def process_label_descriptions(spark, input_data, output_data):
|
||||
""" Parsing label desctiption file to get codes of country, city, state
|
||||
Arguments:
|
||||
spark {object}: SparkSession object
|
||||
input_data {object}: Source S3 endpoint
|
||||
output_data {object}: Target S3 endpoint
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
|
||||
logging.info("Start processing label descriptions")
|
||||
label_file = os.path.join(input_data + "I94_SAS_Labels_Descriptions.SAS")
|
||||
with open(label_file) as f:
|
||||
contents = f.readlines()
|
||||
|
||||
country_code = {}
|
||||
for countries in contents[10:245]:
|
||||
pair = countries.split('=')
|
||||
code, country = pair[0].strip(), pair[1].strip().strip("'")
|
||||
country_code[code] = country
|
||||
|
||||
spark.createDataFrame(country_code.items(), ['code', 'country'])\
|
||||
.write.mode("overwrite")\
|
||||
.parquet(path=output_data + 'country_code')
|
||||
|
||||
city_code = {}
|
||||
for cities in contents[302:962]:
|
||||
pair = cities.split('=')
|
||||
code, city = pair[0].strip("\t").strip().strip("'"),\
|
||||
pair[1].strip('\t').strip().strip("''")
|
||||
city_code[code] = city
|
||||
spark.createDataFrame(city_code.items(), ['code', 'city'])\
|
||||
.write.mode("overwrite")\
|
||||
.parquet(path=output_data + 'city_code')
|
||||
|
||||
state_code = {}
|
||||
for states in contents[981:1036]:
|
||||
pair = states.split('=')
|
||||
code, state = pair[0].strip('\t').strip("'"), pair[1].strip().strip("'")
|
||||
state_code[code] = state
|
||||
spark.createDataFrame(state_code.items(), ['code', 'state'])\
|
||||
.write.mode("overwrite")\
|
||||
.parquet(path=output_data + 'state_code')
|
||||
|
||||
def process_temperature_data(spark, output_data):
|
||||
""" Process temperature data to get dim_temperature table
|
||||
Arguments:
|
||||
spark {object}: SparkSession object
|
||||
input_data {object}: Source S3 endpoint
|
||||
output_data {object}: Target S3 endpoint
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
|
||||
logging.info("Start processing d_temperature")
|
||||
# read temperature data file
|
||||
tempe_data = os.path.join('../../data2/GlobalLandTemperaturesByCity.csv')
|
||||
df = spark.read.csv(tempe_data, header=True)
|
||||
|
||||
df = df.where(df['Country'] == 'United States')
|
||||
d_temperature = df.select(['dt', 'AverageTemperature', 'AverageTemperatureUncertainty',\
|
||||
'City', 'Country']).distinct()
|
||||
|
||||
new_columns = ['dt', 'avg_temp', 'avg_temp_uncertnty', 'city', 'country']
|
||||
d_temperature = rename_columns(d_temperature, new_columns)
|
||||
|
||||
d_temperature = d_temperature.withColumn('dt', to_date(col('dt')))
|
||||
d_temperature = d_temperature.withColumn('year', year(d_temperature['dt']))
|
||||
d_temperature = d_temperature.withColumn('month', month(d_temperature['dt']))
|
||||
|
||||
# write dim_temperature table to parquet files
|
||||
d_temperature.write.mode("overwrite")\
|
||||
.parquet(path=output_data + 'd_temperature')
|
||||
|
||||
def process_demography_data(spark, input_data, output_data):
|
||||
""" Process demograpy data to get dim_demog_population
|
||||
and d_demog_statistics table
|
||||
Arguments:
|
||||
spark {object}: SparkSession object
|
||||
input_data {object}: Source S3 endpoint
|
||||
output_data {object}: Target S3 endpoint
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
|
||||
logging.info("Start processing d_demog_statistics")
|
||||
# read demography data file
|
||||
demog_data = os.path.join(input_data + 'us-cities-demographics.csv')
|
||||
df = spark.read.format('csv').options(header=True, delimiter=';').load(demog_data)
|
||||
|
||||
|
||||
d_demog_statistics = df.select(['City', 'State', 'Male Population', 'Female Population', \
|
||||
'Number of Veterans', 'Foreign-born', 'Race']).distinct() \
|
||||
.withColumn("demog_pop_id", monotonically_increasing_id())
|
||||
|
||||
|
||||
new_columns = ['city', 'state', 'male_population', 'female_population', \
|
||||
'num_vetarans', 'foreign_born', 'race']
|
||||
d_demog_statistics = rename_columns(d_demog_statistics, new_columns)
|
||||
|
||||
# write dim_demog_population table to parquet files
|
||||
d_demog_statistics.write.mode("overwrite")\
|
||||
.parquet(path=output_data + 'd_demog_statistics')
|
||||
|
||||
|
||||
logging.info("Start processing d_demog_statistics")
|
||||
d_demog_statistics = df.select(['City', 'State', 'Median Age', 'Average Household Size'])\
|
||||
.distinct()\
|
||||
.withColumn("d_demog_statistics", monotonically_increasing_id())
|
||||
|
||||
new_columns = ['city', 'state', 'median_age', 'avg_household_size']
|
||||
d_demog_statistics = rename_columns(d_demog_statistics, new_columns)
|
||||
d_demog_statistics = d_demog_statistics.withColumn('city', upper(col('city')))
|
||||
d_demog_statistics = d_demog_statistics.withColumn('state', upper(col('state')))
|
||||
|
||||
# write dim_demog_statistics table to parquet files
|
||||
d_demog_statistics.write.mode("overwrite")\
|
||||
.parquet(path=output_data + 'd_demog_statistics')
|
||||
|
||||
|
||||
'''Paths for local testing'''
|
||||
input_data = "./" # if runs on S3 bucket, please replace by SOURCE_S3_BUCKET
|
||||
output_data = "s3a://gfp-udacity/" # if runs on S3 bucket, please replace by DEST_S3_BUCKET = s3a://gfp-udacity/
|
||||
|
||||
spark = create_spark_session()
|
||||
|
||||
process_immigration_data(spark, output_data)
|
||||
process_label_descriptions(spark, input_data, output_data)
|
||||
process_temperature_data(spark, output_data)
|
||||
process_demography_data(spark, input_data, output_data)
|
||||
1001
data-engineering/06-immigration-capstone/immigration_data_sample.csv
Normal file
1001
data-engineering/06-immigration-capstone/immigration_data_sample.csv
Normal file
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user