Multi-Source Lakehouse Ingestion Pipeline

An end-to-end Python data pipeline that uses FastMCP to expose reusable ingestion tools. It stages mocked PostgreSQL, CSV, Excel, and JSON sales data in Amazon S3, standardizes those sources into a common model, and writes Snappy-compressed Parquet tables to a Silver layer.

Contents

• Architecture

• Features

• Repository structure

• Quick start

• Configuration

• Run the pipeline

• MCP tools

• Data layers and outputs

• Troubleshooting

• Security

• Roadmap

Related MCP server: PySqlitMCP

Architecture

                         BRONZE — RAW INGESTION

 PostgreSQL table ──┐
 CSV file ──────────┼──> FastMCP tools ───> Amazon S3 landing/
 Excel workbook ────┤                         ├── sql/
 JSON API payload ──┘                         ├── csv/
                                            ├── excel/
                                            └── json/
                                                        │
                                                        ▼
                    SILVER — STANDARDIZATION AND PARQUET

                    Source transformers normalize, combine,
                       and deduplicate mocked source data
                                                        │
                                                        ▼
                                             Amazon S3 silver/
                                             ├── fact_sales_transactions/
                                             ├── dim_customers/
                                             └── dim_products/

Features

• Multi-source ingestion: PostgreSQL, CSV, Excel, and JSON inputs.

• MCP tool interface: FastMCP exposes ingestion and transformation routines as callable tools.

• Schema normalization: Source-specific column names are mapped into common sales, customer, and product entities.

• In-memory S3 processing: Landing files are read from S3 and Parquet is written through memory buffers.

• Efficient Silver storage: Output is Apache Parquet with Snappy compression.

• Mocked data included: The repository can be inspected without sharing real business data.

Repository structure

.
├── sample_data/                         # Mocked CSV, Excel, and JSON source files
├── src/
│   ├── config.py                         # Reads AWS and PostgreSQL variables from .env
│   ├── database.py                       # PostgreSQL table-to-CSV extraction
│   ├── storage.py                        # Amazon S3 client and upload helper
│   ├── parquet_utils.py                  # S3 reads and Snappy Parquet writes
│   ├── server.py                         # FastMCP server and public tools
│   └── transformers/
│       ├── sql_source.py                 # PostgreSQL landing-data transformer
│       ├── spreadsheet_source.py         # CSV and Excel transformer
│       ├── json_source.py                # JSON order transformer
│       └── base.py                       # Shared name-cleaning helper
├── test_connections.py                   # Optional AWS and PostgreSQL connectivity test
├── requirements.txt
├── .gitignore
└── README.md

Quick start

Prerequisites

• Python 3.10 or later

• An AWS account, S3 bucket, and IAM credentials with bucket access

• A reachable PostgreSQL instance containing the configured source table

Install

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install -r requirements.txt

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -r requirements.txt

Configuration

Create a .env file in the project root. Never commit this file.

# AWS / S3
AWS_ACCESS_KEY_ID=your_access_key_id
AWS_SECRET_ACCESS_KEY=your_secret_access_key
AWS_REGION=eu-north-1
S3_BUCKET_NAME=your_s3_bucket_name

# PostgreSQL
POSTGRES_HOST=localhost
POSTGRES_PORT=5433
POSTGRES_DB=etl_pipeline
POSTGRES_USER=postgres
POSTGRES_PASSWORD=your_postgres_password

Verify connectivity

python test_connections.py

This tests the configured PostgreSQL connection and lists objects in the configured S3 bucket.

Run the pipeline

1. Stage every source in the Bronze layer

python -c "from src.server import batch_ingest_all_sources; print(batch_ingest_all_sources())"

This stages the mocked local CSV, Excel, and JSON files plus the configured PostgreSQL table under landing/ in S3.

2. Build the Silver layer

python -c "from src.server import build_complete_silver_layer; print(build_complete_silver_layer())"

This reads the landing files, normalizes them, removes duplicates, and writes three Parquet datasets to silver/ in S3.

3. Start the FastMCP server

python src/server.py

An MCP-compatible client can then discover and call the project tools.

MCP tools

Data layers and outputs

Bronze: raw S3 landing files

landing/sql/customer_transactions.csv
landing/csv/source_regional_sales.csv
landing/excel/raw_regional_sales.xlsx
landing/json/source_api_orders.json

Silver: standardized Parquet datasets

silver/fact_sales_transactions/data.parquet
silver/dim_customers/data.parquet
silver/dim_products/data.parquet

The transformations produce these logical entities:

• PostgreSQL column variants such as id / transaction_id and total / amount are handled by the SQL transformer.

• CSV and Excel headers are lowercased, trimmed, and converted to underscore-separated names before their records are combined.

• JSON orders are flattened with pandas.json_normalize().

• Customer names are cleaned into first name, last name, and full name fields.

• Sales are deduplicated by transaction_id plus data_source; customers and products are deduplicated by their IDs.

Troubleshooting

Security

• .env, .env.*, private keys, certificates, and common credential-file names are excluded in .gitignore.

• Do not paste access keys or passwords into source code, issues, commits, or README examples.

• The included sample_data/ files are mocked; replace them with governed data sources for production use.

• For production, use short-lived IAM roles or a managed secret store instead of long-lived local access keys where possible.

Roadmap

• Add automated tests for transformations and S3 integrations.

• Add structured logging, run IDs, and data-quality reports.

• Add a configurable metadata layer for source mappings.

• Add Databricks / Delta Lake Gold-layer transformations.

• Add orchestration through scheduled workflows.

• Add secure deployment configuration and managed secret integration.

License

Add a license file before distributing or reusing this project outside its current scope.