Data Engineering · ETL · Automation

Full-Stack Data Analysis Pipeline

An automated, modular ETL pipeline that ingests raw data from CSV files and REST APIs, transforms and validates it with Pandas, loads it into a PostgreSQL warehouse, and generates scheduled HTML KPI reports via Apache Airflow DAGs.

← Back to Portfolio
Python Pandas PostgreSQL Apache Airflow SQLAlchemy Great Expectations Jinja2 Docker
95%
Time Saved Weekly
<20m
Report Generation
Same Day
Report Turnaround
Auto
Data Quality Checks

Project Overview

A fully automated, four-stage data pipeline designed to eliminate manual reporting bottlenecks. The system ingests raw data from CSV files and REST APIs, applies Pandas transformations with Great Expectations quality validation, persists clean records into a PostgreSQL warehouse via SQLAlchemy ORM, and generates polished HTML KPI dashboards through Jinja2 templates.

Apache Airflow DAGs orchestrate daily runs on a configurable schedule — no manual intervention required. The result: weekly reporting time dropped from 6 hours to under 20 minutes, and data quality issues that previously slipped through undetected are now caught automatically before reaching the warehouse.

Pipeline Architecture

📥

Ingestion

Loads data from CSV files and REST API endpoints into raw DataFrames.

pipeline/ingestion/csv_loader.py · api_loader.py
↓
🔄

Transform & Validate

Pandas cleaning + Great Expectations schema and quality checks. Rejects invalid rows automatically.

pipeline/transform/cleaner.py · validator.py
↓
🗄️

Warehouse

SQLAlchemy ORM loads clean records into PostgreSQL. Declarative models manage schema migrations.

pipeline/warehouse/models.py · loader.py
↓
📊

Reporting

Jinja2 templates render KPI dashboards as standalone HTML files, ready for stakeholder distribution.

pipeline/reporting/generator.py
↓
⏰

Scheduling

Apache Airflow DAG triggers the full pipeline daily. Manual trigger also available via Airflow CLI.

dags/daily_pipeline.py

Impact

MetricBeforeAfter
Weekly reporting time6 hours<20 minutes
Data quality issue detectionManual, ad-hocAutomated (Great Expectations)
Report turnaround2 daysSame day

Quickstart

1. Install dependencies

bash
pip install -r requirements.txt

2. Configure the database

bash
export DATABASE_URL="postgresql://user:password@localhost:5432/analytics"
python -c "from pipeline.warehouse.models import init_db; init_db()"

3. Run the pipeline manually

bash
python -m pipeline.run \
  --input data/sample/sales_sample.csv \
  --output_dir outputs/reports/

4. Schedule with Airflow

bash
# Copy DAG to Airflow home
cp dags/daily_pipeline.py $AIRFLOW_HOME/dags/

# Trigger manually
airflow dags trigger daily_data_pipeline

5. View reports

Open outputs/reports/kpi_report_YYYY-MM-DD.html in a browser to view the generated KPI dashboard.

Stack Details

Ingestion

  • CSV files via Pandas
  • REST API via requests

Transformation

  • Pandas cleaning & shaping
  • Great Expectations validation

Warehouse

  • PostgreSQL 14+
  • SQLAlchemy ORM
  • SQLite for local testing

Reporting & Scheduling

  • Jinja2 HTML templates
  • Apache Airflow 2.7+ DAGs

Project Structure

data-analysis-pipeline/
  ├── pipeline/
    ├── ingestion/
      ├── csv_loader.py
      └── api_loader.py
    ├── transform/
      ├── cleaner.py
      └── validator.py
    ├── warehouse/
      ├── models.py  # SQLAlchemy ORM models
      └── loader.py
    └── reporting/
      ├── generator.py
      └── templates/report.html
  ├── dags/
    └── daily_pipeline.py  # Airflow DAG
  ├── data/sample/
    └── sales_sample.csv
  ├── requirements.txt
  └── README.md