An end-to-end data engineering pipeline built with Apache Airflow to orchestrate data ingestion from Amazon S3 into Amazon Redshift, transform the data into fact and dimension tables, and perform automated data quality validation.
This project demonstrates a complete ETL/ELT data engineering workflow using Apache Airflow, Amazon S3, Amazon Redshift, Python, and SQL.
The pipeline takes raw song and event/log data stored in Amazon S3 and orchestrates the complete data warehouse loading process.
It performs:
- Data staging from Amazon S3
- Redshift table creation
- Fact table loading
- Dimension table loading
- SQL-based transformations
- Automated data quality checks
- Task dependency and workflow management
The main objective is to build a reliable, modular, and automated data pipeline that converts raw data into an analytics-ready warehouse structure.
Amazon S3
│
┌─────────┴─────────┐
│ │
Song Data Event/Log Data
│ │
└─────────┬─────────┘
▼
Apache Airflow
DAG Workflow
│
▼
Create Redshift Tables
│
┌────────┴────────┐
▼ ▼
Stage Songs Stage Events
│ │
└────────┬────────┘
▼
Load Songplays Fact
│
┌────────────┼────────────┐
▼ ▼ ▼
Users Songs Artists
│ │ │
└────────────┼────────────┘
▼
Time
│
▼
Data Quality Checks
│
▼
Pipeline Complete
- 🔄 Automated ETL workflow using Apache Airflow
- ☁️ Amazon S3 ingestion for raw song and event data
- 🏢 Amazon Redshift data warehouse for analytical storage
- 📦 Custom Airflow operators for reusable pipeline tasks
- 🗃️ Staging layer for raw source data
- ⭐ Fact table loading for songplay events
- 👤 Dimension table loading for users
- 🎵 Song dimension processing
- 🎤 Artist dimension processing
- ⏰ Time dimension processing
- ✅ Automated data quality validation
- 🔗 Task dependencies and workflow orchestration
- 🔁 Retry and scheduling support
- 🧩 Reusable SQL query management
The pipeline follows a structured Extract → Transform → Load workflow.
Raw data is stored in Amazon S3.
The project works with two primary types of source data:
S3 Bucket
│
├── song_data/
│
└── log-data/
Song data contains information related to songs and artists, while event/log data represents user activity and song-play events.
Before loading the final warehouse tables, the raw data is loaded into Redshift staging tables.
Amazon S3
│
├── Song Data
│ ↓
│ staging_songs
│
└── Event Data
↓
staging_events
The staging layer separates raw ingestion from final warehouse transformations.
This makes the pipeline easier to debug, maintain, and extend.
Once the source data has been staged in Redshift, SQL queries are used to transform and prepare the data for the final warehouse tables.
The SQL logic is maintained separately inside:
plugins/helpers/sql_queries.py
This keeps the DAG focused on orchestration while keeping transformation logic reusable and maintainable.
The transformed data is loaded into the final fact and dimension tables.
Staging Tables
│
▼
SQL Transformations
│
├── songplays
├── users
├── songs
├── artists
└── time
Apache Airflow is the workflow orchestration engine used in this project.
Airflow is responsible for:
- Scheduling pipeline execution
- Managing task dependencies
- Executing individual tasks
- Handling retries
- Monitoring task status
- Controlling the order of operations
Instead of manually executing every step, Airflow automatically coordinates the entire pipeline.
Create Tables
↓
Stage Data
↓
Load Fact Table
↓
Load Dimensions
↓
Run Quality Checks
↓
Complete
The main workflow is defined in:
dags/udac_example_dag.py
The DAG represents the complete pipeline and defines how individual tasks depend on each other.
Begin Execution
│
▼
Create Redshift Tables
│
├──────────────┐
▼ ▼
Stage Songs Stage Events
│ │
└───────┬──────┘
▼
Load Songplays Fact
│
┌───────┼────────┬────────┐
▼ ▼ ▼ ▼
Users Songs Artists Time
│ │ │ │
└───────┴────────┴────────┘
│
▼
Data Quality Checks
│
▼
Stop Execution
The DAG also uses scheduling, retries, dependencies, and execution controls to make the workflow reliable.
Amazon S3 (Simple Storage Service) is used as the source data storage layer.
The pipeline reads data from an S3 bucket containing:
song_data
log-data
The custom staging operator then loads this data into Redshift staging tables.
S3
│
├── song_data
│ ↓
│ staging_songs
│
└── log-data
↓
staging_events
Amazon Redshift is used as the analytical data warehouse.
The pipeline loads raw data into staging tables and then creates a structured warehouse model consisting of fact and dimension tables.
staging_songs
staging_events
songplays
users
songs
artists
time
The project follows a fact and dimension data warehouse design.
The songplays table represents song-play events.
It acts as the central table for analytical queries.
users
│
│
artists ───── songplays ───── songs
│
│
time
Stores user-related information.
Stores information about songs.
Stores information about artists.
Stores time-related information that can be used for time-based analytics.
This structure allows analytical queries to combine songplay events with descriptive information from the dimensions.
One of the key parts of this project is the implementation of custom Airflow operators.
Located in:
plugins/operators/
The project contains:
create_table.py
stage_redshift.py
load_fact.py
load_dimension.py
data_quality.py
File:
plugins/operators/create_table.py
Responsible for creating the required Redshift tables.
Workflow:
SQL File
↓
CreateTableOperator
↓
PostgresHook
↓
Redshift
↓
Tables Created
File:
plugins/operators/stage_redshift.py
Responsible for loading data from Amazon S3 into Redshift staging tables.
It handles:
- AWS credentials
- S3 bucket
- S3 path
- Target table
- File format
- Redshift connection
Workflow:
Amazon S3
↓
StageToRedshiftOperator
↓
Redshift
↓
Staging Table
File:
plugins/operators/load_fact.py
Responsible for executing SQL that loads data into fact tables.
In this project it is used for loading the:
songplays
fact table.
File:
plugins/operators/load_dimension.py
Responsible for loading dimension tables.
It supports a delete-and-reload operation:
Existing Dimension Data
↓
DELETE
↓
SQL INSERT
↓
Updated Dimension
This operator is reused for:
users
songs
artists
time
File:
plugins/operators/data_quality.py
This operator validates that important warehouse tables contain data.
It checks:
artists
songplays
songs
time
users
The validation uses a simple record-count check:
SELECT COUNT(*) FROM <table>;If the table contains no records, the task fails.
Data Quality Check
│
▼
COUNT(*)
│
┌───────┴───────┐
▼ ▼
Records > 0 Records = 0
│ │
▼ ▼
PASS FAIL
SQL queries are centralized in:
plugins/helpers/sql_queries.py
Instead of writing transformation SQL directly inside the DAG, the project uses reusable SQL query definitions.
Example:
SqlQueries.songplay_table_insertThis provides better separation between:
Workflow Logic
│
▼
DAG
│
▼
Transformation Logic
│
▼
SQL Queries
Benefits:
- Cleaner DAG code
- Reusable SQL
- Easier maintenance
- Better separation of concerns
Airflow dependencies define the order in which tasks execute.
For example:
start_operator >> create_tables_in_redshiftmeans:
Begin
↓
Create Tables
The staging tasks can then execute after table creation:
create_tables_in_redshift >> [
stage_songs_to_redshift,
stage_events_to_redshift
]This allows independent staging operations to run in parallel.
The DAG is configured with scheduling and retry behavior.
Important configurations include:
Retries
Retry Delay
Start Date
Catchup
Maximum Active Runs
Task Dependencies
For example, the pipeline is configured to retry failed tasks:
Task Failure
↓
Wait 5 Minutes
↓
Retry
This is useful when failures are caused by temporary infrastructure or network issues.
Airflow_Data_Pipeline/
│
├── dags/
│ ├── udac_example_dag.py
│ └── sparkify_dimension_subdag.py
│
├── plugins/
│ ├── __init__.py
│ │
│ ├── helpers/
│ │ ├── __init__.py
│ │ └── sql_queries.py
│ │
│ └── operators/
│ ├── __init__.py
│ ├── create_table.py
│ ├── data_quality.py
│ ├── load_dimension.py
│ ├── load_fact.py
│ └── stage_redshift.py
│
├── Setup_Redshift_Connection_Airflow.md
└── README.md
git clone https://github.com/VikasKumar281/Airflow_Data_Pipeline.git
cd Airflow_Data_PipelineInstall and configure Apache Airflow according to your environment.
Make sure Airflow can access:
dags/
plugins/
Create an Airflow connection:
Connection ID:
aws_credentials
This connection is used by the pipeline to access AWS resources.
Create another Airflow connection:
Connection ID:
redshift
Configure the required Redshift/PostgreSQL connection details.
Detailed setup instructions are available in:
Setup_Redshift_Connection_Airflow.md
Start the Airflow services:
airflow schedulerand:
airflow webserverThen:
- Open the Airflow UI.
- Locate the DAG.
- Enable the DAG.
- Trigger it manually or wait for the schedule.
- Monitor task execution.
- Verify the data-quality task.
This project demonstrates practical implementation of:
- Apache Airflow
- DAG orchestration
- ETL / ELT pipelines
- Amazon S3
- Amazon Redshift
- Data staging
- Fact and dimension modeling
- Star-schema concepts
- Custom Airflow operators
- SQL transformations
- Data quality validation
- Task dependencies
- Workflow scheduling
- Retry mechanisms
- AWS integration
- Cloud data warehousing
- Modular pipeline design
Potential improvements include:
- Replace legacy SubDAG implementation with Airflow TaskGroups
- Add more advanced data-quality rules
- Add schema validation
- Add incremental data loading
- Add pipeline monitoring and alerting
- Add automated tests for custom operators
- Add CI/CD using GitHub Actions
- Improve logging and pipeline observability
Contributions are welcome.
git checkout -b feature/your-featureMake your changes, test them, and open a Pull Request.
Developed by Vikas Kumar
GitHub: VikasKumar281
Project:
VikasKumar281/Airflow_Data_Pipeline
Airflow Data Pipeline demonstrates how Apache Airflow can be used to orchestrate a complete cloud data warehouse workflow — from Amazon S3 ingestion and Redshift staging to fact/dimension loading and automated data-quality validation.