Enterprise Data Integration Pipeline

Overview

This project implements a production-style ETL (Extract, Transform, Load) pipeline using Python and SQL concepts. It integrates sales data from two heterogeneous systems (SQL Server and PostgreSQL), applies business transformations and data quality rules, and produces a unified sales fact table and rejection report.

The architecture follows clean OOP principles and is designed to simulate a real-world Snowflake data warehouse ingestion pipeline while remaining fully runnable locally.

---

Architecture

SQL Server CSV        PostgreSQL CSV        Exchange Rates
       │                    │                     │
       └──────┬─────────────┴────────────┬───────┘
              ▼                          ▼
        Extractors (OOP Layer)
              │
              ▼
     Pipeline Transformer (Core Logic)
              │
              ▼
   Data Quality + Business Rules
 (JSON parsing, SKU cleanup, currency conversion,
  duplicate detection, reject handling)
              │
              ▼
   Snowflake Loader (Mock / Interface Layer)
              │
              ▼
   data/processed/
      ├── sales_fact.csv
      └── sales_rejects.csv

---

Project Structure

de-data-integration/
│
├── config/
│   └── dummy.yml
│
├── data/
│   ├── raw/
│   │   ├── sqlserver_mock.csv
│   │   ├── postgres_mock.csv
│   │   └── exchange_rates.csv
│   │
│   └── processed/
│       ├── expected_sales_fact.csv
│       ├── expected_rejects.csv
│       ├── sales_fact.csv
│       └── sales_rejects.csv
│
├── src/
│   ├── extractors/
│   │   ├── base_extractor.py
│   │   ├── sqlserver.py
│   │   └── postgres.py
│   │
│   ├── transformers/
│   │   ├── data_transformer.py
│   │   ├── pipeline_transformer.py
│   │   └── sql_runner.py
│   │
│   ├── loaders/
│   │   └── snowflake.py
│   │
│   └── sql/
│       ├── staging_core.sql
│       ├── staging_acquired.sql
│       └── warehouse_merge.sql
│
├── tests/
│
├── main.py
├── requirements.txt
├── README.md
└── .gitignore

---

Features

• Modular OOP-based ETL pipeline
• Abstract extractor interface for extensibility
• SQL Server and PostgreSQL CSV ingestion
• Snowflake-compatible loader abstraction (mocked locally)
• Currency conversion to USD using exchange rates
• JSON parsing for nested customer data
• SKU normalization and cleanup
• Duplicate transaction detection
• Robust reject handling system
• Full unit test coverage with pytest

---

Data Quality Rules

Core Source

• Invalid JSON in customer_blob
• Missing SKU values

Acquired Source

• Duplicate transaction IDs
• Unknown currency codes

All rejected records are stored in:

data/processed/sales_rejects.csv

---

Transformations

1. SKU Standardization

PROD-9921 → 9921
SKU-8821  → 8821

2. Currency Conversion

All values are normalized to USD using exchange_rates.csv.

3. Customer Parsing

{"id":44,"country":"US"}
→ customer_id = 44, customer_country = US

4. Date Normalization

Handles mixed timestamp formats from different source systems.

5. Warehouse Keying

core:9001
acquired:9001

---

Output

Sales Fact Table

data/processed/sales_fact.csv

Includes:

• Source metadata
• Normalized customer and product data
• Currency-normalized values
• Refund flags

---

Rejected Records

data/processed/sales_rejects.csv

Includes rejected transactions with reasons.

---

Setup Instructions

1. Clone Repository

git clone https://github.com/Habtamariam/de-data-integration.git
cd de-data-integration

---

2. Create Virtual Environment

python -m venv venv

Activate (Windows)

venv\Scripts\activate

Activate (Mac/Linux)

source venv/bin/activate

---

3. Install Dependencies

pip install -r requirements.txt

---

Running the Pipeline

python main.py

Expected Output

Starting Pipeline...
Extraction completed
Transformation completed
Saved 7 rows -> data/processed/sales_fact.csv
Saved 4 rows -> data/processed/sales_rejects.csv
Pipeline finished successfully

---

Running Tests

pytest -v

Expected Result

9 passed

---

Design Decisions

Why OOP Architecture?

The pipeline uses an abstract BaseExtractor to enforce a consistent interface across multiple data sources. This ensures scalability when adding new systems.

---

Why Snowflake Abstraction?

A real Snowflake warehouse is not required. Instead, a mock loader simulates:

• Separation of concerns
• Warehouse compatibility layer
• Easy migration to real Snowflake using COPY INTO or Snowpipe

---

Scalability Strategy

For production-scale datasets (100M+ rows):

• Chunked ingestion
• Parallel processing
• Snowflake COPY INTO staging tables
• Incremental loads
• Partition-based processing

---

Testing Strategy

The project includes unit tests for:

• Extractors (SQL Server & PostgreSQL)
• Transformation logic
• Pipeline orchestration
• Currency conversion
• Reject handling

All tests run locally without external dependencies.

---

Tech Stack

• Python 3.11
• Pandas
• Pytest
• CSV-based data simulation
• OOP design principles
• SQL (Snowflake-compatible design)

---

Notes

This project was built as part of an Enterprise Data Integration take-home assessment.

It emphasizes:

• Clean architecture
• Testability
• Real-world ETL design patterns
• Warehouse-ready thinking (Snowflake-oriented)