Python PostgreSQL Data Quality and ML Evaluation Pipeline

An end-to-end data pipeline built with Python and PostgreSQL to ingest raw CSV data, validate and clean records, detect data-quality issues, and generate analytics and ML evaluation reports.

Key Features

• Ingests raw customer, order, and ML prediction CSV files
• Stores source data in raw PostgreSQL tables
• Validates records using Pydantic and business rules
• Detects missing values, duplicates, invalid emails, negative amounts, invalid churn labels, and unknown customer references
• Loads valid rows into clean analytics-ready tables
• Logs rejected rows into a data quality issues table
• Generates reproducible CSV and JSON reports
• Calculates ML evaluation metrics including accuracy, precision, recall, F1 score, and confusion matrix values
• Includes pytest tests for validation and metric logic

Tech Stack

• Python
• PostgreSQL
• SQLAlchemy
• Pydantic
• Pandas
• Pytest

5. Project Structure

data-quality-ml-pipeline/
│
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── db.py
│   ├── generate_data.py
│   ├── ingest.py
│   ├── metrics.py
│   ├── reports.py
│   ├── run_pipeline.py
│   ├── schema.py
│   └── validate_clean.py
│
├── sql/
│   └── 01_create_tables.sql
│
├── data/
│   ├── raw/
│   └── reports/
│
├── tests/
│   └── test_schema_validation.py
│
├── docker-compose.yml
├── requirements.txt
├── Makefile
├── .env.example
└── README.md

6. What the Pipeline Does in Plain English

A company receives messy data files. Some rows are good. Some rows are broken. This project separates them properly.

The pipeline:

1. Creates sample messy CSV files.
2. Loads all rows into raw PostgreSQL tables.
3. Validates each row using Pydantic and business rules.
4. Sends good rows to clean tables.
5. Sends bad rows to a data-quality issue table.
6. Runs analytics SQL reports.
7. Calculates ML evaluation metrics.
8. Exports reports as CSV and JSON files.

7. Data Sources

The project generates three CSV files:

1. customers.csv
2. orders.csv
3. model_predictions.csv

8. Database Design

The database has four types of tables.

8.1 Pipeline Run Table

pipeline_runs stores one row for every pipeline execution.

Why this matters:

If something fails, you can trace which raw rows and validation issues belong to that exact run.

8.2 Raw Tables

Raw tables store data exactly as received from CSV files.

• raw_customers
• raw_orders
• raw_model_predictions

Bad data is allowed here because raw storage should preserve the original input.

8.3 Clean Tables

Clean tables store only validated rows.

• clean_customers
• clean_orders
• clean_model_predictions

8.4 Issue Table

data_quality_issues stores rejected rows and failure reasons.

Example issue types:

• PYDANTIC_VALIDATION_ERROR
• DUPLICATE_CUSTOMER_ID
• DUPLICATE_ORDER_ID
• UNKNOWN_CUSTOMER_ID
• INVALID_COUNTRY

9. Validation Checks

The pipeline detects:

• Missing email values
• Invalid email format
• Invalid age values
• Duplicate customer IDs
• Duplicate order IDs
• Negative order amounts
• Orders linked to unknown customers
• Predictions linked to unknown customers
• Invalid churn labels
• Invalid prediction scores
• Unsupported country values

10. ML Evaluation Metrics

The project calculates:

• Accuracy
• Precision
• Recall
• F1 score
• True positives
• True negatives
• False positives
• False negatives

In this project:

• actual_churn = 1 means the customer really churned.
• predicted_churn = 1 means the model predicted churn.

11. Setup Instructions

11.1 Start PostgreSQL

docker compose up -d

Check that the database container is running:

docker ps

11.2 Create Python Virtual Environment

python3 -m venv venv
source venv/bin/activate

On Windows:

venv\Scripts\activate

11.3 Install Dependencies

pip install -r requirements.txt

11.4 Configure Environment

The project already includes a sample .env file for local Docker usage.

If needed, copy .env.example to .env:

cp .env.example .env

12. Run the Full Pipeline

python -m app.run_pipeline

Expected output:

Creating database tables...
Executed SQL file: sql/01_create_tables.sql
Generating sample data...
Sample raw CSV files generated successfully.
Creating pipeline run...
Pipeline run_id: 1
Ingesting raw files...
Ingested 7 rows into raw_customers
Ingested 6 rows into raw_orders
Ingested 6 rows into raw_model_predictions
Validating and cleaning data...
Validated customers. Clean rows inserted: 3
Validated orders. Clean rows inserted: 3
Validated predictions. Clean rows inserted: 3
Generating reports...
JSON report created: data/reports/pipeline_report_run_1.json
CSV report created: data/reports/data_quality_issues_summary_run_1.csv
CSV report created: data/reports/customer_analytics_run_1.csv
CSV report created: data/reports/category_revenue_run_1.csv
CSV report created: data/reports/ml_prediction_details_run_1.csv
Pipeline completed successfully.

13. Run Tests

pytest -q

The tests check core validation behavior:

• valid customer rows pass
• invalid email fails
• negative order amount fails
• invalid churn label fails
• safe metric division works

14. Generated Reports

After running the pipeline, reports are created inside:

data/reports/

Reports:

pipeline_report_run_1.json
data_quality_issues_summary_run_1.csv
customer_analytics_run_1.csv
category_revenue_run_1.csv
ml_prediction_details_run_1.csv

15. Example JSON Report Content

The JSON report contains:

{
    "project_name": "Python PostgreSQL Data Quality and ML Evaluation Pipeline",
    "run_id": 1,
    "assumptions": [
        "customer_id must be unique in the customers file",
        "order_id must be unique in the orders file",
        "orders must reference an existing clean customer",
        "prediction rows must reference an existing clean customer"
    ],
    "table_counts": {
        "raw_customers": 7,
        "raw_orders": 6,
        "raw_model_predictions": 6,
        "clean_customers": 3,
        "clean_orders": 3,
        "clean_model_predictions": 3,
        "data_quality_issues": 10
    },
    "ml_evaluation_metrics": {
        "total_predictions": 3,
        "true_positive": 1,
        "true_negative": 1,
        "false_positive": 1,
        "false_negative": 0,
        "accuracy": 0.6667,
        "precision": 0.5,
        "recall": 1.0,
        "f1_score": 0.6667
    }
}

16. How to Explain This Project in an Interview

You can say:

I built a Python and PostgreSQL pipeline that simulates a real analytics data workflow. The pipeline ingests raw CSV files for customers, orders, and ML predictions. I intentionally included bad data such as duplicates, missing values, invalid IDs, invalid churn labels, unsupported country values, and negative order amounts.

The data first lands in raw PostgreSQL tables. Then Python validation logic using Pydantic checks each row. Good rows are loaded into clean analytical tables, while bad rows are stored in a data_quality_issues table with the exact issue type and reason.

After that, I run SQL-based analytics reports and ML evaluation reports. The project generates reproducible CSV and JSON reports containing assumptions, row counts, data-quality metrics, and model-performance metrics like accuracy, precision, recall, F1 score, and confusion matrix values.

17. GitHub Push Commands

git init
git add .
git commit -m "Initial commit: data quality and ML evaluation pipeline"
git branch -M main
git remote add origin https://github.com/YOUR_USERNAME/data-quality-ml-pipeline.git
git push -u origin main

18. Suggested Commit History

If you want cleaner commits:

git init

git add docker-compose.yml requirements.txt .env.example .gitignore Makefile
git commit -m "Add project setup and dependencies"

git add sql/01_create_tables.sql app/config.py app/db.py
git commit -m "Add PostgreSQL schema and database utilities"

git add app/generate_data.py app/ingest.py
git commit -m "Add sample data generation and raw ingestion"

git add app/schema.py app/validate_clean.py
git commit -m "Add data validation and cleaning pipeline"

git add app/metrics.py app/reports.py
git commit -m "Add reporting and ML evaluation metrics"

git add app/run_pipeline.py tests/ README.md
git commit -m "Add pipeline runner, tests, and documentation"

19. Important Notes

This project intentionally uses a simple architecture.

That is the point.

It proves the fundamentals clearly:

• schema design
• raw versus clean tables
• validation logic
• business-rule checks
• SQL analytics
• reproducible reports
• ML metric calculation
• failure-case documentation

Before adding heavy tools like Airflow, Spark, or Kafka, this basic version should work correctly. Complexity without correctness is just expensive confusion wearing a hoodie.