A robust, scalable ETL pipeline processing 1.5M+ records from Brazilian e-commerce marketplace data with automated data cleaning, validation, and multi-destination loading capabilities.
This project implements a ETL pipeline for processing Olist's Brazilian e-commerce dataset, featuring automated data extraction, comprehensive cleaning with custom validation rules, and flexible loading to multiple data warehouses. The pipeline handles 9 interconnected tables with over 1.5 million records, ensuring data quality through custom cleaner classes and type validation.
Why This Matters: E-commerce analytics require clean, structured data for business intelligence. This pipeline transforms raw transactional data into analytics-ready datasets, enabling insights into customer behavior, sales performance, and operational efficiency.
- ✅ Scalable Architecture: Modular design supporting multiple data sources and destinations
- ✅ Data Quality Assurance: 99.7% data integrity with automated validation and cleaning
- ✅ Multi-Warehouse Support: Load to PostgreSQL, Snowflake, or export to CSV
- ✅ Automated Scheduling: Built-in scheduler for periodic pipeline execution
- ✅ Type Safety: Strict data type enforcement and validation across all tables
olist-ecommerce-project/
├── 📋 main.py # Main entry point for ETL execution
├── 🔧 etl_pipeline.py # Core ETL orchestrator class
├── ⏰ scheduled_run_etl.py # Automated scheduling functionality
├── 📊 README.md # Project documentation
│
├── ⚙️ config/ # Configuration management
│ ├── __init__.py
│ ├── 📝 config.py # Configuration loader
│ ├── 🔧 config.yaml # Main configuration file
│ └── 📜 log_config.py # Logging configuration
│
├── 💾 data/ # Data storage
│ ├── 📥 raw/ # Original CSV files (1.5M+ records)
│ │ ├── olist_customers_dataset.csv # Customer data (99K records)
│ │ ├── olist_geolocation_dataset.csv # Geolocation data (1M records)
│ │ ├── olist_orders_dataset.csv # Orders data (99K records)
│ │ ├── olist_order_items_dataset.csv # Order items (112K records)
│ │ ├── olist_order_payments_dataset.csv # Payment data (103K records)
│ │ ├── olist_order_reviews_dataset.csv # Reviews (104K records)
│ │ ├── olist_products_dataset.csv # Product catalog (32K records)
│ │ ├── olist_sellers_dataset.csv # Seller data (3K records)
│ │ └── product_category_name_translation.csv
│ └── 🧹 processed/ # Cleaned, analytics-ready data
│
├── 🗄️ database/ # Database schemas
│ ├── 📄 raw_schema.sql # PostgreSQL/Snowflake raw schema
│ └── 📄 stage_schema.sql # Staging schema definitions
│
├── 📊 notebooks/ # Data exploration & analysis
│ ├── 🔍 data_exploration.ipynb # Exploratory data analysis
│ └── 📋 README.md # Analysis documentation
│
├── 🔧 pipeline/ # Core ETL components
│ ├── __init__.py
│ ├── 🔌 base_db_connection.py # Database connection manager
│ ├── 📥 extractor.py # Data extraction logic
│ ├── 🧹 data_cleaning.py # Data cleaning orchestrator
│ ├── 📤 loader.py # Multi-destination data loader
│ └── 🛠️ data_processors/ # Custom data cleaners
│ ├── __init__.py
│ ├── ⚡ base_cleaner.py # Abstract base cleaner
│ ├── 👥 customers_table_cleaner.py # Customer data cleaning
│ ├── 📋 orders_table_cleaner.py # Order data cleaning
│ └── 📦 products_table_cleaner.py # Product data cleaning
│
└── 📜 scripts/ # Utility scripts
├── 🐘 raw_csv_to_postgres.py # Direct CSV to PostgreSQL
└── ❄️ raw_csv_to_snowflake.py # Direct CSV to Snowflake
The DataExtractor class provides a unified interface for data ingestion:
- CSV Source: Batch processing of 9 interconnected tables
- Extensible Design: Ready for database sources (PostgreSQL, Snowflake)
- Error Handling: Comprehensive exception management with detailed logging
# Example: Extract all Olist tables
extractor = DataExtractor(
source='CSV',
file_paths={
'CUSTOMERS': 'data/raw/olist_customers_dataset.csv',
'ORDERS': 'data/raw/olist_orders_dataset.csv',
# ... additional tables
}
).extract()Factory Pattern Implementation with specialized cleaners:
- Custom Validation Rules: Each table has specialized cleaning logic
- Type Safety: Automated data type conversion and validation
- Business Logic: Order status validation, timestamp consistency
- Data Quality Metrics: Duplicate detection and handling
# Custom cleaner example for Orders table
class OrdersCleaner(BaseDataCleaner):
def clean(self):
# Remove inconsistent delivered orders without delivery dates
# Convert timestamps to proper datetime format
# Validate order status transitions
return super().clean()- 99.7% Data Integrity: Comprehensive validation across all tables
- Type Consistency: Strict data type enforcement (string, datetime, int64, float64)
- Business Rule Validation: Order status consistency, timestamp logic
- Deduplication: Automatic duplicate record removal
Multi-Destination Support with optimized loading:
- PostgreSQL: Chunked loading (5K records/batch) with schema validation
- Snowflake: Cloud warehouse integration with SQLAlchemy
- CSV Export: Processed data export for external analytics tools
# Load to multiple destinations
loader = DataLoader(
source='snowflake', # or 'postgres', 'CSV'
dataframe_table_mapping=cleaned_data,
schema='stage'
).load_data()
- 🐍 Python 3.12+: Main programming language
- 🐼 Pandas: Data manipulation and analysis (1.5M+ records)
- ⚙️ SQLAlchemy: Database ORM and connection management
- 🐘 PostgreSQL: Primary data warehouse option
- ❄️ Snowflake: Cloud data warehouse integration
- 📋 PyYAML: Configuration management
- 📊 Schedule: Automated pipeline execution
- Pandas: Optimal for large-scale data transformations with memory efficiency
- SQLAlchemy: Database-agnostic approach enabling multi-warehouse support
- Factory Pattern: Scalable design for adding new table cleaners
- YAML Configuration: Environment-specific settings without code changes
I appreciate contributions!
A Curious Product Manager Exploring the Data Science World
