Multi-Source Lakehouse Pipeline

An end-to-end, orchestrated data pipeline that uses FastMCP to ingest mocked PostgreSQL, CSV, Excel, and JSON sales data into an AWS S3 landing zone, standardizes it into a common model as Parquet, builds it up through a Databricks (Unity Catalog + Delta Lake) lakehouse into business-ready Gold tables, and serves those tables two ways: a 5-page Databricks AI/BI dashboard and a Genie Space for natural-language Q&A. A Databricks Job chains the full lakehouse build and dashboard refresh into a single, repeatable, on-demand or scheduled run — from raw files to a query-ready agent, following each step automatically.

Contents

• Architecture

• Features

• Repository structure

• Quick start

• Configuration

• Run the pipeline

• MCP tools

• Data layers and outputs

• AI/BI dashboard

• Orchestration

• Genie Space

• 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

                    Python transformers normalize, combine,
                       and deduplicate mocked source data
                                                        │
                                                        ▼
                                             Amazon S3 silver/
                                             ├── fact_sales_transactions/
                                             ├── dim_customers/
                                             └── dim_products/
                                                        │
                                                        ▼
   ┌─────────────────────────────────────────────────────────────────┐
   │      DATABRICKS JOB: MCP_Data_Pipeline_Run (orchestration)       │
   │                                                                   │
   │  Setup_CatalogSchema → Load_StagingData → Clean_StagingData →     │
   │              Creating_GoldLayer → Customer_Sales_Report           │
   │                                                                   │
   │      DATABRICKS LAKEHOUSE — UNITY CATALOG (mcp_lakehouse)         │
   │                                                                   │
   │  silver_staging (external Delta tables over S3 silver/ Parquet)   │
   │                          │                                       │
   │                          ▼                                       │
   │      silver_clean (managed Delta tables — deduped,                │
   │       surrogate keys, standardized types & currency)              │
   │                          │                                       │
   │                          ▼                                       │
   │             gold (business-ready Delta marts)                    │
   │             ├── customer_segmentation                            │
   │             ├── product_performance                              │
   │             ├── regional_sales                                   │
   │             ├── regional_product_performance                     │
   │             └── monthly_sales                                    │
   └─────────────────────────┬─────────────────────────┬─────────────┘
                              │                         │
                              ▼                         ▼
      DATABRICKS AI/BI DASHBOARD              GENIE SPACE (natural-
        (Customer Sales Report)                 language Q&A over
   Executive Summary · Regional Analysis         the Gold tables)
   Product Performance · Customer Analytics
              · Global Filters

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.

• Lakehouse layer on Databricks: Unity Catalog (mcp_lakehouse) with silver_staging, silver_clean, and gold schemas built on Delta Lake.

• Business-ready Gold marts: PySpark aggregations for customer segmentation (RFM-style recency/frequency/monetary with Active/Risky/Inactive status), product performance, and regional/monthly sales.

• Databricks AI/BI dashboard: A 5-page Lakeview dashboard (Customer_Sales_Report.lvdash.json) built directly on the Gold tables, with KPI counters, trend and regional charts, and an interactively sortable top-customers table.

• End-to-end orchestration: A Databricks Job (MCP_Data_Pipeline_Run) chains catalog setup, staging, cleaning, Gold-layer build, and dashboard refresh into a single on-demand or scheduled run, with Unity Catalog lineage tracked automatically.

• Genie Space: A natural-language analytics agent over the Gold tables, letting non-technical users ask sales, customer, and regional questions directly without writing SQL.

• 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
├── DDL/
│   ├── all_schemas_access.sql             # Catalog + schema bootstrap (silver_staging, silver_clean, gold)
│   ├── silver_staging/                    # External Delta table definitions over S3 silver/ Parquet
│   ├── silver_clean/                      # Managed Delta table definitions (cleaned, keyed)
│   └── gold/                              # Business mart table definitions
├── Notebooks/
│   ├── 1_Creating mcp pipeline catalog and schema.ipynb   # Unity Catalog + external table setup
│   ├── 2_Loading and viewing staging silver schema data.ipynb  # Sanity-check silver_staging
│   ├── 3_Cleaning and Transforming silver staging and loading in silver clean.ipynb
│   └── 4_Business Ready Gold Layer.ipynb  # Builds all 5 Gold tables
├── Customer_Analytical_Report/
│   └── Customer_Sales_Report.lvdash.json  # Databricks AI/BI (Lakeview) dashboard definition
├── assets/
│   └── screenshots/                       # README screenshots (orchestration Job, Genie Space)
├── 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

• A Databricks workspace with Unity Catalog enabled, and access to the same S3 bucket (for the lakehouse layer, dashboard, orchestration Job, and Genie Space)

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 (S3 Parquet)

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.

4. Build the Databricks lakehouse layer

Run the following notebooks, in order, against a Databricks cluster or SQL warehouse with Unity Catalog access:

The DDL under DDL/ mirrors what these notebooks create and can be run directly (e.g. DDL/all_schemas_access.sql) to bootstrap or inspect the catalog independently of the notebooks.

5. Import the AI/BI dashboard

In a Databricks workspace with access to the mcp_lakehouse.gold schema, import Customer_Analytical_Report/Customer_Sales_Report.lvdash.json as a new Lakeview dashboard (Dashboards → Create → Import). See AI/BI dashboard below for what it contains.

6. Automate steps 4–5 with the orchestration Job

Steps 4 and 5 don't have to be run by hand every time — the MCP_Data_Pipeline_Run Databricks Job chains catalog setup, staging, cleaning, the Gold build, and the dashboard refresh into one on-demand or scheduled run. See Orchestration below.

7. Ask questions in the Genie Space

Once the Gold tables exist, the MCP Data Pipeline Sales Analytics Agent Genie Space can answer sales, customer, and regional questions directly in natural language. See Genie Space below.

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 (S3)

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

silver_staging: external Delta tables (Databricks, mcp_lakehouse.silver_staging)

Read directly off the S3 silver/ Parquet output — same shape as above, with the addition of data_source for lineage back to the originating system.

silver_clean: cleaned, keyed Delta tables (Databricks, mcp_lakehouse.silver_clean)

gold: business-ready Delta marts (Databricks, mcp_lakehouse.gold)

• 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.

• In silver_clean, only postgresql_legacy amounts are converted from INR to USD — other sources are already in USD and pass through unchanged. Surrogate keys (*_key) are added, and invalid rows (non-positive amount/quantity, missing customer ID) are dropped before the Gold layer is built.

AI/BI dashboard

Customer_Analytical_Report/Customer_Sales_Report.lvdash.json is a Databricks AI/BI (Lakeview) dashboard built directly on the 5 Gold tables. It ships as 5 pages:

The Customer Analytics page's top-customers table is parameter-driven rather than statically sorted: sort_by (Monetary Value / Purchase Frequency) and sort_order (Highest to Lowest / Lowest to Highest) filter widgets are bound to SQL CASE expressions in the underlying query, so the ranking updates live as a viewer changes the filters — no static "top N" assumption baked into the query.

Orchestration

MCP_Data_Pipeline_Run is a Databricks Job that chains the entire lakehouse build into a single run, in dependency order:

Setup_CatalogSchema → Load_StagingData → Clean_StagingData → Creating_GoldLayer → Customer_Sales_Report

Unity Catalog tracks lineage across the whole run automatically — the job currently reports 13 upstream tables and 11 downstream tables — and Photon performance optimization is enabled on the job cluster. The job can be triggered on demand (Run now) or put on a recurring schedule from the same Jobs & Pipelines UI.

Genie Space

MCP Data Pipeline Sales Analytics Agent is a Databricks Genie Space built directly on the Gold tables, letting anyone — not just SQL users — ask sales, customer, and regional questions in plain English and get answered from governed, business-ready data. Capabilities include:

• Identifying high-value customers via RFM segmentation, lifetime value, and activity status.

• Tracking monthly sales trends, revenue, customer counts, and product quantities for forecasting and year-over-year comparisons.

• Analyzing product-level performance (customer reach, units sold, revenue) to inform portfolio decisions.

• Comparing product performance and sales across regions for distribution and marketing planning.

Example questions it answers directly: "Give me Top Customers data who has higher monetary value based on their status," "In which region there was maximum number of transactions?", and "What is the monthly trend of total sales revenue?"

Because it sits on top of the same Gold tables the dashboard and orchestration Job maintain, the Genie Space always reflects the latest successful pipeline run — no separate data prep required.

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 data-quality checks in silver_clean before Gold aggregation runs.

• Add a configurable metadata layer for source mappings.

• Add secure deployment configuration and managed secret integration.

• Expand Genie Space instructions/examples to cover product- and region-level drill-downs.

License

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