Udacity-Data-Pipeline-with-Airflow

Udacity Data Engineering Nanodegree Program, Data Pipeline with Airflow project using MinIO and Postgresql.

Project Structure:

• dags: Directory containing Airflow DAG scripts.

  • create_tables_and_buckets.py: DAG for creating tables on PostgreSQL and buckets on MinIO.
  • create_tables.sql: SQL script containing CREATE TABLE queries.
  • pipeline_dag.py: Main DAG script for the ETL data pipeline.

• data: Directory for storing project source data.

  • log_data: Subdirectory for log data.
  • song_data: Subdirectory for song data.

• plugins: Directory for custom Airflow plugins.

  • operators: Subdirectory for operator scripts.

    • init.py: Initialization script for operators.
    • stage_postgresql_operator.py: Operator script for copying data from MinIO to PostgreSQL.
    • load_fact_operator.py: Operator script for executing INSERT queries into the fact table.
    • load_dimension_operator.py: Operator script for executing INSERT queries into dimension tables.
    • data_quality_operator.py: Operator script for performing data quality checks.

  • helpers: Subdirectory for helper scripts.

    • init.py: Initialization script for helpers.
    • sql_queries.py: Script containing SQL queries for building dimensional tables.

Airflow Dag Overview

Staging & Fact Table Schema

Dimension Tables Schema

How to Run

Clone the project repository

$ git clone https://github.com/akarce/Udacity-Data-Pipeline-with-Airflow

Change directory

$ cd Udacity-Data-Pipeline-with-Airflow

Build the docker images and start airflow:

$ docker-compose up airflow-init $ docker compose up -d

Create database called sparkifydb:

$ docker exec -it postgresdev psql -U postgres_user -d postgres_db

$ CREATE DATABASE sparkifydb;

WebUI links: \

Airflow webserver http://localhost:8080/

MinIO http://localhost:9001/

Sign in to Airflow webserver:

Username: airflow Password: airflow

Create Postgres connection:

Go to admin -> Connections -> Add a new record

Connection Id: postgres_conn
Connection Type: Postgres
Host: 172.18.0.1
Database: sparkifydb
Login: postgres_user
Password: postgres_password
Port: 5432

Obtain Access Key from MinIO:

Go to MinIO WebUI using http://localhost:9001/
From Access Keys section -> create access key -> Create
Store you Access Key and Secret Key using Download for Import

Create MinIo connection:

Go to admin -> Connections -> Add a new record

Connection Id: minio_conn
Connection Type: Amazon Web Services
AWS Access Key ID: <your_key_id>
AWS Secret Access Key: <your_secret_key> \

Trigger the create_tables_and_buckets_dag using Airflow WebUI

This will create two buckets named
udacity-dend, processed
and 7 postgres tables named
artists, songplays, songs, staging_events, staging_songs, times, users

To see the schemas of tables use \

$ \d <table_name>

Upload the log_data and song_data folders from data folder into udacity-dend bucket from MinIO WebUI

Username: minioadmin
Password: minioadmin

Finally unpause the dag named pipeline:

After finished successfully you can check the logs for data quality messages, and the all objects will be moved to the processed bucket.

You can see the Airflow logs for data quality results:

=======

1b712ebd7a3d8fd64ccb00116f6c053a5f10cde7