🌬️ Fragrance Enrichment Pipeline

A modular, auditable data pipeline built with Apache Airflow and Docker, designed to ingest, clean, and enrich fragrance metadata from personal collections and external sources like Fragrantica. This project emphasizes traceability, data governance, and recruiter-friendly documentation.

🎯 Project Context

Challenge: Learn Apache Airflow and modern data orchestration through hands-on implementation rather than tutorials.

Approach: Built complete fragrance metadata enrichment pipeline combining personal collection tracking with web-scraped data from Fragrantica.

Outcome: Production-ready Dockerized platform with data governance, comprehensive testing, and full documentation.

Timeline: Self-paced learning project (2024) - approximately 40-60 hours of development

Learning Method: Self-taught through documentation, trial and error, and AI-assisted peer review (Co-pilot, Claude). No courses - just real problem-solving.

📊 Project Metrics

DAGs Developed: 3 production pipelines
Python ETL Scripts: 3 applications (load_fragrances, load_fragrantica, load_customers)
Container Services: 2 (Airflow + PostgreSQL)
Data Sources: Excel, CSV (multiple encodings/delimiters handled)
Development Approach: Self-taught, documentation-driven, AI peer-reviewed
Status: Production-ready, running in local Docker environment
Lines of Code: ~500+ Python, 100+ YAML/Dockerfile

📁 Repository Structure

airflow-docker/
├── audit/                     # Governance artifacts and audit logs
│   └── audit_log.md
├── dags/                      # Airflow DAGs and pipeline logic
│   ├── data/                  # Raw or staging data inputs
│   ├── load/                  # Source ingestion scripts
│   ├── test/                  # DAG and utility test scripts
│   └── transform/             # Enrichment and transformation DAGs
├── docs/                      # Data quality and governance documentation
│   └── data_quality.md
├── logs/                      # Execution logs and traceability
├── Dockerfile                 # Airflow image with SQL Server support
├── docker-compose.yaml        # Multi-container orchestration
├── Fragrance_Project_data.txt # Sample enrichment logic
├── license.md                 # Licensing and usage terms
└── .gitignore                 # Git exclusions

🏗️ Architecture

┌─────────────────────────────────────────┐
│   Airflow Webserver (8080)              │
│   Airflow Scheduler                     │
├─────────────────────────────────────────┤
│   PostgreSQL Metadata DB (5432)        │
│   - Health Checks                       │
│   - DAG History                         │
├─────────────────────────────────────────┤
│   Volumes:                              │
│   - ./dags → /opt/airflow/dags         │
│   - ./logs → /opt/airflow/logs         │
│   - postgres-db-volume (persistent)    │
└─────────────────────────────────────────┘
         ↓
    SQL Server 2022
    (host.docker.internal:2383)
         ↓
    [fragrance_collection]
    [FragranticaData_temp]

🛠️ Pipeline Overview

This project orchestrates three primary ETL flows:

1. Personal Fragrance Collection Loader

Source: Excel file (Stephen Stephan Fragrance Collection.xlsx)
Process: Deduplication, data cleaning
Target: SQL Server fragrance_collection table
Post-processing: Archives original file with timestamp

2. Fragrantica Metadata Loader

Source: CSV file (fra_cleaned.csv)
Process: Rating normalization, casing standardization, note field parsing
Target: SQL Server FragranticaData_temp table (append mode)
Post-processing: Deduplicates by url, archives source file

3. Customer Data Loader

Source: Excel file (H+ Sport Customers.xlsx)
Process: Deduplication, PII removal (Zipcode column)
Target: SQL Server customers table
Post-processing: Archives original file with timestamp

📅 Airflow DAGs

`etl_load_fragrance_collection`

Sensor: Waits for Excel file to appear (FileSensor with 60s polling, 3-hour timeout)
ETL Task: Executes load_fragrances.main()
Archive: Moves processed file to timestamped archive folder

`etl_load_fragrantica`

ETL Task: Executes load_fragrantica.main()
Cleanup: Normalizes ratings, applies title casing, deduplicates by URL
Archive: Moves CSV to timestamped archive folder

`etl_load_customers`

Sensor: Waits for Excel file to appear
ETL Task: Executes load_customers.main()
Data Governance: Removes PII (Zipcode) before loading
Archive: Moves processed file to timestamped folder

📄 DAG Scheduling

All DAGs use schedule_interval=None for manual triggering
catchup=False to prevent backfills of historical runs
Tagged with etl, sqlserver, and fragrances for UI filtering
Manual execution allows for controlled data ingestion and validation

🧼 ETL Logic: Personal Fragrance Collection

Script: dags/load/load_fragrances.py

Process:

Loads Excel file using pandas.read_excel()
Drops duplicate rows based on all columns
Connects to SQL Server using SQLAlchemy + pyodbc with environment-based credentials
Writes to fragrance_collection table with if_exists="replace" strategy
Archives original file with timestamp to ./archive directory

Key Code:

# Secure credential management
params = urllib.parse.quote_plus(
    f"DRIVER={{ODBC Driver 17 for SQL Server}};"
    f"SERVER={os.environ.get('SQL_SERVER')};"
    f"DATABASE={os.environ.get('SQL_DB')};"
    f"UID={os.environ.get('SQL_USER')};"
    f"PWD={os.environ.get('SQL_PASSWORD')}"
)
engine = create_engine(f"mssql+pyodbc:///?odbc_connect={params}")

# Load to SQL Server
fragrances.to_sql("fragrance_collection", engine, if_exists="replace", index=False)

# Timestamped archiving
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
archive_path = os.path.join(archive_dir, f"fragrance_collection_{timestamp}.xlsx")
shutil.move(excel_file, archive_path)

🌐 ETL Logic: Fragrantica Metadata

Script: dags/load/load_fragrantica.py

Complex Transformations:

Reads CSV with semicolon delimiter and ISO-8859-1 encoding (handles accented characters in French perfume names)
Normalizes rating format: European decimal (4,5) → US decimal (4.5) with type conversion
String normalization: Replaces hyphens with spaces, applies title case to Brand/Perfume/Gender

Comma-separated list processing for Top/Middle/Base notes:

def titlecase_notes(series):
    return series.fillna('').apply(
        lambda x: ', '.join([note.strip().title() for note in x.split(',')])
    )

Multi-column iteration: Cleans 5 main accord columns with lambda functions
URL-based deduplication with audit logging of removed records
Appends to FragranticaData_temp table for staging

Key Code:

# Encoding and delimiter handling
df = pd.read_csv(csv_file, delimiter=';', encoding='ISO-8859-1')

# Rating normalization
df['Rating Value'] = df['Rating Value'].str.replace(',', '.').astype(float)

# String transformations
df['Perfume'] = df['Perfume'].str.replace('-', ' ').str.title()
df['Brand'] = df['Brand'].str.replace('-', ' ').str.title()

# Deduplication with logging
initial_count = df.shape[0]
df = df.drop_duplicates(subset='url', keep='first')
print(f"Removed {initial_count - df.shape[0]} duplicate rows based on 'url'")

# Append mode for staging
df.to_sql("FragranticaData_temp", engine, if_exists="append", index=False)

Data Quality Note:

# Identified failed and alleged cleaned data from the fragrantica scraped file.
# Cleaned my ass!

Reality check: "Cleaned" data required extensive normalization for production use.

🛡️ Governance Highlights

Data Quality Controls

All transformations logged in audit/audit_log.md
Original files archived for rollback and forensic review
Data quality checklist maintained in docs/data_quality.md
.gitignore excludes sensitive and large files (.csv, .xlsx, /archive/, /audit/)

Audit Trail Example

Every ingestion event tracked with:

Timestamp
Source file name
Rows ingested
Duplicates removed
Transformation notes

Security Practices

Environment-based credential management (no hardcoded passwords)
SQL Server credentials via environment variables
ODBC connection string encoding for special characters
Separate .env file (excluded from version control)

🳠Docker & Airflow Setup

This project uses Docker to containerize Airflow and its dependencies, including support for SQL Server via ODBC.

🔧 Dockerfile Highlights

Based on apache/airflow:2.8.1
Installs Microsoft ODBC Driver 17 for SQL Server connectivity
Adds Python dependencies: pandas, sqlalchemy, pyodbc, openpyxl

FROM apache/airflow:2.8.1

USER root
# Install ODBC Driver 17 for SQL Server
RUN apt-get update && apt-get install -y curl gnupg && \
    curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \
    curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \
    apt-get update && ACCEPT_EULA=Y apt-get install -y msodbcsql17

USER airflow
RUN pip install --no-cache-dir pandas sqlalchemy pyodbc openpyxl

🧩 docker-compose.yaml

Defines two services:

1. Postgres Service (Airflow Metadata DB)

Health checks for reliable startup
Named volume for data persistence
Exposed on port 5432

2. Airflow Service (Scheduler + Webserver)

Depends on healthy Postgres service
Environment variables for SQL Server connectivity
Mounted volumes for DAGs and logs (cached for performance)
Initialization script creates admin user and DB schema

Key Configuration:

environment:
  AIRFLOW__CORE__EXECUTOR: LocalExecutor
  AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
  AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
  SQL_SERVER: ${SQL_SERVER}
  SQL_DB: ${SQL_DB}
  SQL_USER: ${SQL_USER}
  SQL_PASSWORD: ${SQL_PASSWORD}

volumes:
  - ./dags:/opt/airflow/dags:cached
  - ./logs:/opt/airflow/logs:cached
  - postgres-db-volume:/var/lib/postgresql/data

🧾 Audit Logging

All ingestion events are tracked in audit/audit_log.md:

Timestamp	Source File	Rows Ingested	Duplicates Removed	Notes
2025-08-10 18:45 ET	fragrantica_dump.csv	12,348	1,203	Rating format normalized (comma → dot)
2025-08-09 17:12 ET	basenotes_raw.csv	8,902	842	Gender field inconsistencies resolved
2025-08-08 14:30 ET	personal_collection.xlsx	247	12	Duplicate perfumes removed

📊 Data Quality Checklist

Outlined in docs/data_quality.md, covering:

Pre-Ingestion Validation:

✅ Encoding and delimiter validation
✅ Required column presence checks
✅ File format verification

Transformation Quality:

✅ Null handling and imputation strategies
✅ Deduplication logic by unique identifiers
✅ Format normalization (dates, numbers, text)
✅ Semantic cleanup (title casing, special characters)

Future Enhancements:

📋 Schema validation with pandera or great_expectations
📋 Fuzzy matching for cross-source brand/perfume joins
📋 Outlier detection and anomaly logging
📋 Automated data quality scoring

🚀 Getting Started

1. Clone the Repository

git clone https://github.com/SStephanJX/fragrance-enrichment-pipeline.git
cd airflow-docker

2. Configure Environment Variables

# Create .env file with SQL Server credentials
cat > .env << EOF
SQL_SERVER=your_server_address
SQL_DB=your_database_name
SQL_USER=your_username
SQL_PASSWORD=your_password
EOF

3. Build and Start the Containers

docker-compose up --build

This will:

Build the Airflow image with SQL Server ODBC support
Start Postgres and Airflow services with health checks
Initialize the Airflow database schema
Create an admin user (airflow:airflow)
Mount DAG and log directories

4. Access the Airflow UI

Open your browser and navigate to:

http://localhost:8080

Login credentials:

Username: airflow
Password: airflow

5. Trigger a DAG

Upload required data files to dags/data/:
- Stephen Stephan Fragrance Collection.xlsx
- fra_cleaned.csv
- H+ Sport Customers.xlsx
Navigate to DAGs page in Airflow UI
Toggle the DAG to "On" state
Click "Trigger DAG" button (play icon)
Monitor task status and logs in real-time

🔧 Troubleshooting

DAGs not appearing in UI?

✅ Check ./dags volume mount in docker-compose.yaml
✅ Verify DAG files have no Python syntax errors
✅ Review Airflow logs: docker-compose logs airflow
✅ Confirm DAG files are in correct directory structure

SQL Server connection failing?

✅ Verify environment variables in docker-compose.yaml
✅ Confirm SQL Server allows remote connections
✅ Check firewall rules for port 1433
✅ Validate ODBC Driver 17 installation: docker exec -it airflow odbcinst -q -d

File sensor timing out?

✅ Ensure files are in dags/data/ directory
✅ Check filename matches exactly (case-sensitive)
✅ Verify file permissions allow reading
✅ Increase timeout value in DAG configuration if needed

Permission errors on logs/dags?

✅ Set proper ownership: sudo chown -R $(id -u):$(id -g) ./dags ./logs
✅ Verify Docker has read/write access to mounted volumes
✅ Check SELinux policies if on RHEL/CentOS

🛣️ Roadmap

Planned enhancements include:

Phase 1: Data Quality

✅ Incremental loading for large datasets (avoid full reloads)
✅ Schema validation with pandera or great_expectations
📋 Automated anomaly detection and alerting

Phase 2: Integration

✅ Cross-source joins with fuzzy matching (brand name variations)
📋 API integration for real-time fragrance data
📋 Webhook triggers for automated DAG execution

Phase 3: Visualization

📋 Dashboarding via Metabase or Streamlit
📋 Data lineage visualization
📋 Quality metrics dashboard

Phase 4: Scale

📋 Migration to cloud-based Airflow (AWS MWAA, GCP Composer)
📋 Distributed executor for parallel processing
📋 Data partitioning strategies for large datasets

🧠 What This Demonstrates

This isn't a tutorial project — it's a complete production-grade data engineering platform built from scratch through self-directed learning.

Technical Skills

✅ Apache Airflow Orchestration - DAG design, file sensors, Python operators (self-taught, no courses)
✅ Docker Infrastructure - Multi-container orchestration, health checks, volume persistence, network configuration
✅ Python ETL Development - pandas DataFrames, SQLAlchemy ORM, complex transformations, lambda functions
✅ SQL Server Integration - ODBC connectivity, secure credential management, environment-based configuration
✅ Data Quality Engineering - Encoding normalization, deduplication, multi-column transformations, audit logging
✅ Complex String Processing - Comma-separated list parsing, title casing, delimiter handling, encoding issues

Engineering Practices

✅ Production Patterns - Error handling, retry logic, file archiving, comprehensive logging
✅ Infrastructure as Code - docker-compose.yaml, Dockerfile, repeatable deployments
✅ Documentation First - Every component documented for maintainability and knowledge transfer
✅ Data Governance - Audit trails, quality checks, rollback capability, PII handling
✅ Security Best Practices - Environment variables, credential encoding, .gitignore exclusions
✅ Version Control - Proper .gitignore, structured commits, clean repository

Learning Approach

✅ Self-Directed - Built through Apache Airflow documentation, Docker guides, not tutorials
✅ AI-Assisted - Used Co-pilot and Claude as peer reviewers, not code generators
✅ Problem-Solving - Trial, error, and iteration methodology when stuck
✅ Production Focus - Built with enterprise patterns from day one, not prototypes

💼 Recruiter-Facing Summary

This project demonstrates:

ETL Mastery

SQL Server integration via ODBC with secure credential handling
Apache Airflow orchestration with file sensing and manual triggering
Docker containerization with multi-service orchestration
Python pandas for complex data transformations

Governance-First Mindset

Comprehensive audit logs for every data transformation
Data quality checklists and validation procedures
File archiving for rollback and forensic capabilities
PII handling and data security considerations

Documentation Rigor

Every pipeline step is traceable and reproducible
README provides complete setup and troubleshooting
Code comments explain complex transformation logic
Architecture diagrams show system design

Professional Branding

Built to showcase technical excellence and ethical boundaries
Portfolio-grade code quality and structure
Modern development practices (Docker, Git, AI-assisted)
Production-ready patterns throughout

🔗 Related Projects

This Airflow platform is part of a comprehensive data engineering portfolio:

1. Fragrance Data Warehouse (SCENTED_DW)

Dimensional star schema consuming staged data from this pipeline for analytics.

Technologies: SQL Server, Dimensional Modeling, Star Schema, Bridge Tables, Kimball Methodology
Features: Fact tables, dimension tables, many-to-many relationships via bridge tables

2. Employment Data Warehouse

Separate dimensional modeling project for career analytics and skills tracking.

Technologies: SQL Server, Dimensional Design, Fact/Dimension Architecture
Features: Skills matrix, role tracking, course completion analytics

3. HRIS Analytics System

Production Power BI dashboard built in 23 hours for take-home project.

Technologies: Power BI, Python, Flask, SQLite, pytest
Features: Hiring metrics, departmental analytics, time-to-hire calculations, brand-aligned colors

4. E-Commerce Platform

Production Shopify site with Jinja2 templating and catalog optimization.

Technologies: Shopify Liquid (Jinja2), HTML/CSS/JavaScript, Java (initial prototype)
Features: Custom templates, dynamic content, live customer traffic

Portfolio: GitHub Profile | LinkedIn

🙌 Acknowledgments

External Data Sources:

Fragrantica (web-scraped fragrance metadata)
Basenotes (fragrance community data)
Personal collection tracking data

Technology Stack:

Apache Airflow container base: apache/airflow:2.8.1
SQL Server connectivity: Microsoft ODBC Driver 17
PostgreSQL for Airflow metadata storage
Docker and Docker Compose for containerization

Learning Resources:

Apache Airflow official documentation
Docker and Docker Compose documentation
Python pandas and SQLAlchemy documentation
AI-assisted peer review (Co-pilot, Claude)

All third-party content is used for educational and analytical purposes only.

🎓 Learning Outcomes

This project was built entirely through self-directed learning, documentation research, and AI-assisted peer review. It demonstrates:

✅ The ability to architect and deploy a complete data engineering platform without formal instruction
✅ Mastery of orchestration, transformation, and infrastructure layers through hands-on implementation
✅ Production-grade patterns including credential safety, archival, and audit logging learned through iteration
✅ Problem-solving through trial, error, and documentation — not tutorials or courses
✅ Modern development practices including AI-assisted peer review as a productivity tool
✅ Self-motivation and initiative to learn emerging technologies independently

Key Learning: Building real systems with production patterns teaches more than any course. This project represents approximately 40-60 hours of focused learning, problem-solving, and implementation — all self-directed.

📜 License & Usage

Portfolio Project: This code is shared publicly for demonstration and educational purposes.

Personal Use: Enrichment logic and data pipeline patterns developed for personal learning and skill development.

External Data: Fragrantica and other datasets used under fair use for educational analysis only. Not intended for commercial redistribution.

Code Reference: Architecture patterns and approaches may be referenced for learning purposes. Please cite this repository if using substantial portions in your own work.

Author: Stephen Stephan
Contact: LinkedIn | Email | GitHub

Last Updated: January 2025

Name		Name	Last commit message	Last commit date
dags		dags
docs		docs
.gitignore		.gitignore
Dockerfile		Dockerfile
Fragrance_Project_data.txt		Fragrance_Project_data.txt
docker-compose.yaml		docker-compose.yaml
license.md		license.md
readme.md		readme.md

Latest commit

History

Folders and files

Repository files navigation