US E-commerce retail data pipeline - transaction and inventory analysis

This project was created as part of the finale for the DataTalksClub - Data Engineering Zoomcamp 6 week course. The data used for this project is completely artifical; they were generated using Python scripts. There are 4 tables in total:

1. Two dimensional tables which contains product and store data
2. Two fact tables which contain inventories and transactions data. In order to create a dashboard, some historical data is generated starting Jan 2024 - March 2024. The Airflow job can be run every hour or so to simulate the generation of transactions and inventories every few seconds. The number of products and stores is config driven and is set to 2_000 and 1_000 records respectively.

Problem Description

The goal of the project is to build an end to end streaming data pipeline that will produce data, between 5 and 10 seconds, for a fictitious US E-commerce retail giant. The dimensional data and fact data are store in Google Cloud Storage. Pubsublite is used to produce and consume streaming transactions and inventories, which is processed using Apache Spark; the final results are stored in Bigquery, and they are transformed using dbt; the analysis is displayed on Streamlit.

Technology Stack

The following technologies have been used

• Google BigQuery
• Google Cloud Storage (GCS)
• Google Dataproc
• Google Pubsublite
• Terraform
• Airflow
• Apache Spark
• dbt
• Docker
• Make
• Pipenv
• Streamlit

Data Dictionary

Schema for product

Field name	Type	Description
product_id	STRING	Unique identifier
name	STRING	Product name
category	STRING	Product classification
base_price	FLOAT	Unit price
supplier_id	STRING	Unique supplier identifer

Schema for store

Field name	Type	Description
store_id	STRING	Unique identifier
location	STRING	Store location
size	INT	Store size
manager	STRING	Name of manager

Schema for transaction

Field name	Type	Description
transaction_id	STRING	Unique identifier
product_id	STRING	Product identifier
timestamp	STRING	Time of transaction
quantity	INT	Number of units
unit_price	FLOAT	Price of product
store_id	STRING	Store identifer

Schema for inventory

Field name	Type	Description
inventory_id	STRING	Unique identifier
product_id	STRING	Product identifier
timestamp	STRING	Time of transaction
quantity_change	INT	Change in product quantity
store_id	STRING	Store identifer

High level architecture

End to end data flow chart

Streamlit demo: US retail dashboard

Click

The data used for this demo is artifically generated using Python scripts

Steps to reproduce the project

1. Prerequisites:

Google Cloud Platform Account

Note - If you have already done these steps then it is not required.

• Sign up for a free account here, and enable billing.
• Create your project
• Create a service account under IAM & Admin
• Grant the following roles - Owner + Storage Admin + Storage Object Admin + Service Usage Admin
• Once done, under Actions against your Service Account, click on Manage keys. Click Add key, and then Create new key. Download the JSON file and store it in a suitable location locally
• Enable the following apis - serviceusage here

Google Cloud SDK - Optional

Installation instruction here.

Enable Google authentication - Optional

export GOOGLE_APPLICATION_CREDENTIALS=<path/to/your/service-account-authkeys>.json
gcloud auth activate-service-account --key-file $GOOGLE_APPLICATION_CREDENTIALS
gcloud auth application-default login

Install Docker for your OS

Installation instructionshere

Install Terraform

You can view the installation instructions for Terraform here

2. Clone the repository:

git clone https://github.com/sl2902/retail_data_pipeline.git

3. Change the working directory:

cd retail_data_pipeline/

4. Rename the env.template file to .env:

mv env.template .env

4.1 Fill in the blanks to the following environment variables in the .env file and save them:

project_id=
bucket_name=
dataset=
SERVICE_ACCOUNT_FILENAME=
HOST_GOOGLE_APPLICATION_CREDENTIALS=/path/to/your/${SERVICE_ACCOUNT_FILENAME}

5. Build and enable the virtual environment:

make setup

6. Start Docker desktop and run docker compose:

make docker_build

7. Load environment variables into the project directory:

source .env

8. Create GCP Storage bucket and enable required Cloud APIs using Terraform:

cd terraform

8.1 Initialize terraform:

terraform init

8.2 Plan terraform:

terraform plan

8.3 Apply terraform:

terraform apply

Enter "yes" when prompted for a value to the following "Do you want to perform these actions?" Note - On some occassions, the following api cloudresourcemanager. However, on checking the console, the API appears enabled. In such cases ignore the error. Otherwise, manually enable it via the console.

8.4 Return to project root folder:

cd ../

9. Launch Airflow UI. username - airflow. password - airflow:

http://localhost:8080

9.1 Run sanity check to see whether the dags are available; there should be 5 dags, which are paused:

make dag_list

Note - if you haven't provided the SERVICE_ACCOUNT_FILENAME and project_id, it will fail to add the connection

9.2 Create service account in Airflow connections:

make add_gcp_service_account_airflow

9.3 Unpause all dags:

make dag_unpause_all_dags

Note: Sometimes it takes a while to unpause the DAGs. Run the make dag_list command to check whether the paused field is set to False. You can also manually unpause the dags from the Airflow UI home page; it is located to the left of the dag name.

9.4 Run the retail data pipeline dag:

make dag_run_retail_data_pipeline

9.4.1 Check the status of job either on the CLI or via the Airflow UI by clicking on the respective dag:

make dag_chk_status_run_retail_data_pipeline

10. Run the streamlit dashboard:

make run_streamlit

Note - Because you may not have updated the secrets file required by Streamlit to connect to Bigquery, you may see a message on the dashboard, you can ignore it as the script will fall back on the service account file defined in the .env file. If you want to avoid the warning message, then you will need to create a .streamlit folder in the project directory and a corresponding secrets.toml with the following contents drawn from your gcp service account file

[gcp_service_account]

type = "service_account"
project_id = "PROJECT_ID"
private_key_id = "KEY_ID"
private_key = "-----BEGIN PRIVATE KEY-----\nPRIVATE_KEY\n-----END PRIVATE KEY-----"
client_email = "SERVICE_ACCOUNT_EMAIL"
client_id = "CLIENT_ID"
auth_uri = "https://accounts.google.com/o/oauth2/auth"
token_uri = "https://accounts.google.com/o/oauth2/token"
auth_provider_x509_cert_url = "https://www.googleapis.com/oauth2/v1/certs"
client_x509_cert_url = "https://www.googleapis.com/robot/v1/metadata/x509/SERVICE_ACCOUNT_EMAIL"

11. Simulate generating transactions and inventories in real-time and stream the data for 5 seconds:

make dag_run_publish_stream_to_bq_sec

11.1 Check the status of job either on the CLI or via the Airflow UI by clicking on the respective dag:

make dag_chk_status_publish_stream_to_bq_sec

11.2 Check the updated streamlit dashboard

make run_streamlit

11.3 Clean the environment

make docker_clean

12. Destroy terraform resources:

cd terraform && terraform destroy && cd ../

12.1 Remove the remaining dependent APIs either via the console or after installing gcloud sdk and initializing the project:

  gcloud services list --format="value(config.name)" \
| xargs -I{} gcloud services disable --force {}

References

[1] Pub/Sub Lite Spark Connector
[2] Airflow Docker
[3] dbt Materialized View
[4] Add GCP Credentials via Airfolw CLI
[5] Disable all GCP APIs via CLI
[6] Enable service account roles using Terraform
[7] Using Airflow decorators instead of Jinja