Build a Star Schema Data Warehouse

Model a Baltic FMCG warehouse with Python ETL and SCD Type 2 history.

Introduction

30 Second Summary

Every growing company eventually hits a wall where data lives in separate systems across different offices or countries. Ask a simple question like "what was our total revenue last quarter?" and you get a different answer depending on which system you pull from.

In this project, you will build a star schema data warehouse on SQL Server that consolidates sales data from three simulated country ERPs. Your Python ETL pipeline will handle deduplication, change tracking, and incremental loads to produce trusted cross-border reporting.

What You'll Build

You will run a cross-country sales query against your finished warehouse and see clean, trustworthy revenue numbers that the same query against the raw flat data cannot produce.

By the end of this project, you'll have:

  • A flat staging table loaded with three countries' ERP exports, where you can see firsthand how duplicates and missing history make reporting unreliable.
  • A star schema with DimProduct, DimCustomer (SCD Type 2 history tracking), DimSubsidiary, DimDate, and FactSalesOrder that produces correct cross-border revenue in a single query.
  • A Python ETL pipeline using SQLAlchemy and pyodbc that you run end-to-end to transform messy source data into properly dimensioned facts with resolved surrogate keys.
  • Secret Mission: Process a second batch with changed customer segments and verify your SCD Type 2 logic correctly preserves history while updating current records.

Are there any prerequisites?

No prior experience with data warehousing or SQL Server is required. Step 1 walks you through installing every tool from scratch on Windows.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters for real-world data engineering.

Install Your Data Engineering Toolkit

A data warehouse needs three core pieces: a database engine to store the warehouse, a query tool to inspect it, and a scripting language to run the ETL pipeline that feeds it.

In this step, you will install SQL Server 2025 as your local database, SSMS 22 as your query tool, and Python 3.13 with the libraries that connect them.

In this step, get ready to:
  • Install SQL Server 2025 Developer Edition as your local database engine.
  • Install SSMS 22 and the ODBC driver that lets Python talk to SQL Server.
  • Set up Python 3.13 with a virtual environment and project dependencies.
Install SQL Server 2025 Developer Edition

SQL Server 2025 Developer Edition is free and includes all Enterprise features for development and testing. You will use it as the engine that hosts your star schema warehouse.

  • Open your browser and navigate to the SQL Server downloads page.
  • Scroll down to the SQL Server 2025 Developer section.
  • Click Download Enterprise Developer edition.
  • Run the downloaded installer.

What happens when you run the installer?

The installer presents three options: Basic, Custom, and Download Media. Basic performs a straightforward installation of the SQL Engine with sensible defaults. That is what you want here.

  • Select Basic.
  • Accept the license terms when prompted.
  • Leave the install location at its default and click Install.
  • Wait for the installation to complete. You will see a summary screen confirming that the SQL Server Database Engine installed successfully.

What is SQL Server Developer Edition?

Developer Edition gives you the full SQL Server Enterprise feature set at zero cost. The only restriction is that you cannot use it for production workloads. For learning data warehousing locally, it is the ideal choice.

🙋‍♀️ Installation stuck or failing?

Make sure you are running the installer as an administrator. Right-click the installer file and select Run as administrator.

If you see a firewall warning, you can safely ignore it. A firewall rule is only needed if other machines need to connect to your SQL Server.

Why can't I install SQL Server 2025?

Install SSMS 22 and ODBC Driver 18

SSMS (SQL Server Management Studio) is the graphical tool you will use to write and run queries against your warehouse. ODBC Driver 18 is the bridge that lets Python's pyodbc library communicate with SQL Server.

  • Download SSMS 22 by opening this link in your browser:
https://aka.ms/ssms/22/release/vs_SSMS.exe
  • Run the downloaded vs_SSMS.exe file.
  • The Visual Studio Installer will open. Click Install to begin.
  • Wait for the installation to finish. Close the installer when prompted.

Why does SSMS use the Visual Studio Installer?

Starting with SSMS 22, Microsoft distributes SSMS through the Visual Studio Installer rather than a standalone MSI. The stub file vs_SSMS.exe bootstraps that installer and handles the download for you.

Now install the ODBC driver that Python will use to connect.

  • Download the ODBC Driver 18 x64 MSI by opening this link in your browser:
https://go.microsoft.com/fwlink/?linkid=2358430
  • Run the downloaded .msi file.
  • Click through the installer, accepting the license agreement and leaving defaults.
  • Click Install, then Finish when complete.

ODBC installer asks about Visual C++ Redistributable?

ODBC Driver 18 requires the Visual C++ Redistributable. Most Windows systems already have it. If prompted, allow the installer to install it automatically.

Help me install ODBC Driver 18 for SQL Server

Set up Python 3.13 and create the project

Python runs the ETL pipeline that will transform raw CSV data into your star schema. You need version 3.13 specifically because the pinned library versions have pre-built wheels for it.

  • Press the Windows key to open your search bar.
  • Type PowerShell and press Enter to open it.
  • Check whether Python is already installed by running:
python --version

✔️ I see version 3.13 or higher

Python 3.13 is ready. Skip ahead to the project folder step below.

ⓧ I see an older version

You have Python installed but at an older version. The libraries this project uses require 3.13.

  • Open https://www.python.org/downloads/windows/ in your browser.
  • Download the Windows installer (64-bit) for Python 3.13.
  • Run the installer.
  • On the first screen, check the box labelled Add Python to PATH. This is unchecked by default and is critical for the commands in this project to work.
  • Click Install Now.
  • Close and reopen PowerShell after installation so the new PATH takes effect.
  • Run python --version again and confirm it now shows 3.13.

ⓧ Command not found

Python is not installed yet.

  • Open https://www.python.org/downloads/windows/ in your browser.
  • Download the Windows installer (64-bit) for Python 3.13.
  • Run the installer.
  • On the first screen, check the box labelled Add Python to PATH. This is unchecked by default and is critical for the commands in this project to work.
  • Click Install Now.
  • Close and reopen PowerShell after installation so the new PATH takes effect.
  • Run python --version again and confirm it shows 3.13.

Why check that box?

The Add Python to PATH option tells Windows where to find the python and pip commands. Without it, every terminal command in this project would fail with a "not recognized" error.

Now create your project folder. You will keep all files for this project here.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir baltic-dw
cd baltic-dw
  • Create a Python virtual environment by running:
python -m venv .venv

What is a virtual environment?

A virtual environment is an isolated copy of Python where you install project-specific packages. It prevents version conflicts between projects on your machine.

  • Activate the virtual environment by running:
.venv\Scripts\activate

Your terminal prompt now starts with (.venv) confirming the virtual environment is active.

  • Install the three libraries your ETL pipeline needs by running:
pip install pyodbc==5.3.0 sqlalchemy==2.0.51 pandas==3.0.5

What do these libraries do?

  • pyodbc is the low-level driver that sends SQL commands from Python to SQL Server via ODBC.
  • sqlalchemy provides the create_engine pattern and enables fast_executemany for bulk inserts.
  • pandas handles reading CSV files and pushing DataFrames into SQL Server via to_sql.

You should see output ending with "Successfully installed" followed by the three package names.

Pip install fails?

Make sure your virtual environment is active. You should see (.venv) at the start of your prompt. If not, run .venv\Scripts\activate again.

If you see a permissions error, do not use sudo or --user. Instead confirm you are inside the baltic-dw folder with the .venv directory present.

Help me fix my pip install in this Python virtual environment

Finally, open VS Code in this folder so you have an editor ready for the next steps.

  • Open VS Code in the current folder by running:
code .

VS Code launches with the baltic-dw folder loaded in the Explorer sidebar.

Verify your full setup

Time to confirm that SQL Server and Python can both do their jobs.

  • Press the Windows key to open your search bar.
  • Type SSMS and press Enter to open SQL Server Management Studio.
  • In the Connect to Server dialog, confirm the Server name is set to localhost and Authentication is Windows Authentication.
  • Click Connect.

Before you run the next query, think about what you expect to see. What version of SQL Server did you just install?

  • Click New Query in the toolbar, paste the following, and press F5 to execute:
SELECT @@VERSION

The results pane shows a row containing "Microsoft SQL Server 2025" with version and build details. This confirms your database engine is running.

Now verify the Python side. Switch back to your PowerShell terminal (the one with the (.venv) prompt).

  • Confirm pyodbc is installed and can be imported by running:
python -c "import pyodbc; print(pyodbc.version)"

You should see 5.3.0 printed. This proves Python can load the ODBC library that will connect to your SQL Server.

SSMS won't connect to localhost?

Verify that the SQL Server service is running. Press the Windows key, type services.msc, and press Enter. Look for SQL Server (MSSQLSERVER) in the list. Its status should be Running. If it shows Stopped, right-click it and select Start.

Help me connect SSMS to my local SQL Server 2025 instance

Your entire data engineering toolkit is installed and verified. Next up, you will generate realistic multi-country sales data and load it into a flat table to see exactly why a naive approach breaks down.

Generate Multi-Country ERP Data and Load It Flat

Your toolkit is installed and ready. Now you need data that represents the real-world problem: three Baltic subsidiaries each running a separate ERP system that produces its own sales orders, customer records, and product catalogs.

In this step, you will generate that multi-country data, dump it all into one flat table, and see firsthand why a naive "just merge everything" approach produces wrong numbers. This is the problem that motivates a proper star schema.

In this step, get ready to:
  • Generate simulated ERP exports for Estonia, Latvia, and Lithuania as CSV files.
  • Load all 150 rows flat into a SQL Server staging table.
  • Query the flat data and observe why it breaks for cross-border reporting.
Create the data generation script

Each Baltic subsidiary exports orders in its own format with its own customer IDs. You will write a Python script that simulates these three ERP exports.

  • In VS Code, create a new file called generate_source_data.py in your project folder.
  • Add the imports, configuration, and product catalog by pasting this code:
import pandas as pd
import os
from pathlib import Path
from datetime import datetime, timedelta
import random

random.seed(42)

# Resolve paths relative to this script, not the terminal's working directory
PROJECT_DIR = Path(__file__).resolve().parent
OUTPUT_DIR = PROJECT_DIR / "source_data"
os.makedirs(OUTPUT_DIR, exist_ok=True)

# Product catalog (shared across countries with slight naming variations)
products = [
    {"product_id": "P001", "product_name": "Colgate Total Toothpaste 100ml", "category": "Oral Care", "brand": "Colgate-Palmolive"},
    {"product_id": "P002", "product_name": "Palmolive Shower Gel 250ml", "category": "Personal Care", "brand": "Colgate-Palmolive"},
    {"product_id": "P003", "product_name": "Zewa Deluxe Toilet Paper 8pk", "category": "Tissue", "brand": "Essity"},
    {"product_id": "P004", "product_name": "Libresse Normal Pads 20pk", "category": "Hygiene", "brand": "Essity"},
    {"product_id": "P005", "product_name": "Dilmah Ceylon Tea 100bags", "category": "Beverages", "brand": "Dilmah"},
    {"product_id": "P006", "product_name": "Shell Helix Ultra 5W-40 1L", "category": "Lubricants", "brand": "Shell"},
    {"product_id": "P007", "product_name": "Shell Rimula R6 10W-40 5L", "category": "Lubricants", "brand": "Shell"},
    {"product_id": "P008", "product_name": "Zewa Just-1 Kitchen Towel 2pk", "category": "Tissue", "brand": "Essity"},
]

What does this code do?

This sets a fixed random seed so the generated data is reproducible every time you run it. The product catalog represents the FMCG brands a Baltic distributor would carry.

Path(__file__).resolve().parent computes the folder this script lives in, so OUTPUT_DIR always resolves correctly no matter where your terminal is when you run the script. Without it, a bare "source_data" string would look for the folder relative to your terminal's current directory, which breaks if you run from anywhere else.

  • Below the products list, add the customer lists for all three countries:
# Customers per country (some appear in multiple countries)
customers_ee = [
    {"customer_id": "EE-C001", "customer_name": "Selver AS", "segment": "Retail", "city": "Tallinn"},
    {"customer_id": "EE-C002", "customer_name": "Rimi Eesti", "segment": "Retail", "city": "Tartu"},
    {"customer_id": "EE-C003", "customer_name": "Tallinn Hotels Group", "segment": "HORECA", "city": "Tallinn"},
    {"customer_id": "EE-C004", "customer_name": "Baltic Fuel OÜ", "segment": "Standard", "city": "Pärnu"},
    {"customer_id": "EE-C005", "customer_name": "Grossi Toidukaubad", "segment": "Retail", "city": "Tallinn"},
]

customers_lv = [
    {"customer_id": "LV-C001", "customer_name": "Maxima Latvija", "segment": "Retail", "city": "Riga"},
    {"customer_id": "LV-C002", "customer_name": "Rimi Latvia", "segment": "Retail", "city": "Riga"},
    {"customer_id": "LV-C003", "customer_name": "Lido Leisure Group", "segment": "HORECA", "city": "Riga"},
    {"customer_id": "LV-C004", "customer_name": "Baltic Fuel SIA", "segment": "Standard", "city": "Liepaja"},
    {"customer_id": "LV-C005", "customer_name": "Mego Stores", "segment": "Retail", "city": "Daugavpils"},
]

customers_lt = [
    {"customer_id": "LT-C001", "customer_name": "Maxima LT", "segment": "Retail", "city": "Vilnius"},
    {"customer_id": "LT-C002", "customer_name": "Rimi Lietuva", "segment": "Retail", "city": "Kaunas"},
    {"customer_id": "LT-C003", "customer_name": "Amberton Hotels", "segment": "HORECA", "city": "Vilnius"},
    {"customer_id": "LT-C004", "customer_name": "Baltic Fuel UAB", "segment": "Standard", "city": "Klaipeda"},
    {"customer_id": "LT-C005", "customer_name": "Norfa Stores", "segment": "Retail", "city": "Vilnius"},
]

Why do some customers appear in multiple countries?

Notice that "Baltic Fuel" operates in all three countries with different legal entity suffixes (OÜ, SIA, UAB). Each country ERP assigns its own customer ID. This is the real-world consolidation problem you will solve later with your star schema.

  • Below the customer lists, add the order generation function:
def generate_orders(country_code, customers, start_order_id, num_orders=50):
    orders = []
    base_date = datetime(2024, 1, 15)

    for i in range(num_orders):
        order_id = f"{country_code}-ORD-{start_order_id + i:05d}"
        order_date = base_date + timedelta(days=random.randint(0, 350))
        customer = random.choice(customers)
        product = random.choice(products)
        quantity = random.randint(5, 200)
        unit_price = round(random.uniform(2.50, 45.00), 2)

        orders.append({
            "order_id": order_id,
            "order_date": order_date.strftime("%Y-%m-%d"),
            "customer_id": customer["customer_id"],
            "customer_name": customer["customer_name"],
            "customer_segment": customer["segment"],
            "customer_city": customer["city"],
            "product_id": product["product_id"],
            "product_name": product["product_name"],
            "product_category": product["category"],
            "product_brand": product["brand"],
            "quantity": quantity,
            "unit_price": unit_price,
            "line_total": round(quantity * unit_price, 2),
            "country_code": country_code,
        })

    return pd.DataFrame(orders)

What does this function do?

Each call generates 50 order lines for one country. Every order row is fully denormalized: customer info, product info, and order measures are all flattened into a single row. This mirrors how real ERP exports look before anyone models them properly.

  • Finally, add the execution code that generates and saves all three country files:
df_ee = generate_orders("EE", customers_ee, 10000, 50)
df_lv = generate_orders("LV", customers_lv, 20000, 50)
df_lt = generate_orders("LT", customers_lt, 30000, 50)

df_ee.to_csv(os.path.join(OUTPUT_DIR, "estonia_orders.csv"), index=False)
df_lv.to_csv(os.path.join(OUTPUT_DIR, "latvia_orders.csv"), index=False)
df_lt.to_csv(os.path.join(OUTPUT_DIR, "lithuania_orders.csv"), index=False)

print(f"Generated {len(df_ee)} Estonia orders")
print(f"Generated {len(df_lv)} Latvia orders")
print(f"Generated {len(df_lt)} Lithuania orders")
print(f"Files saved to {OUTPUT_DIR}/")

What does this code do?

Each country gets a different starting order ID range (10000, 20000, 30000) so order IDs never collide. The three DataFrames are saved as separate CSVs, simulating what you would receive from each country's ERP system.

  • Save generate_source_data.py.
  • Run the script to generate your source data:
python generate_source_data.py

You should see output confirming 50 orders generated for each country and files saved to the source_data/ directory.

  • Confirm the CSV files exist by running:
ls source_data/

You should see estonia_orders.csv, latvia_orders.csv, and lithuania_orders.csv listed.

Not seeing the CSV files?

  • Make sure you saved generate_source_data.py before running it.
  • Check that your terminal is in the project folder (the same directory as generate_source_data.py).
  • If you see a ModuleNotFoundError for pandas, confirm your virtual environment is activated.

Why aren't my CSV files being generated?

✔️ Awesome, I've got everything!

Great. Double check you have saved generate_source_data.py and move on to the next substep.

ⓧ I'd like to double check the full code

Scroll up and confirm each of these sections is present in your file, in order:

  • Imports and configuration (random seed, OUTPUT_DIR, makedirs).
  • Products list (8 items).
  • Three customer lists (customers_ee, customers_lv, customers_lt) with 5 entries each.
  • The generate_orders function.
  • Three generate_orders calls, three to_csv calls, and four print statements.
Create the database and load the data flat

Now you need a SQL Server database to receive this data. You will create a staging table that stores every column from the CSVs in one wide, flat table.

  • Switch back to SSMS. Click New Query in the toolbar to open a new query window.
  • Paste the following SQL to create the database and staging table:
-- Create the BalticDW database
IF NOT EXISTS (SELECT name FROM sys.databases WHERE name = 'BalticDW')
    CREATE DATABASE BalticDW;
GO

USE BalticDW;
GO

-- Staging table (flat, denormalized)
IF OBJECT_ID('stg_raw_orders', 'U') IS NOT NULL DROP TABLE stg_raw_orders;
CREATE TABLE stg_raw_orders (
    order_id NVARCHAR(20),
    order_date DATE,
    customer_id NVARCHAR(20),
    customer_name NVARCHAR(100),
    customer_segment NVARCHAR(50),
    customer_city NVARCHAR(100),
    product_id NVARCHAR(10),
    product_name NVARCHAR(200),
    product_category NVARCHAR(50),
    product_brand NVARCHAR(50),
    quantity INT,
    unit_price DECIMAL(10,2),
    line_total DECIMAL(12,2),
    country_code NVARCHAR(5)
);
GO

What does this SQL do?

  • Creates a BalticDW database if it does not already exist.
  • Creates stg_raw_orders as a flat staging table with every column from the CSVs. No primary keys, no foreign keys, no normalization. This is intentionally naive.
  • Click Execute (or press F5) to run the query.
  • Confirm it succeeded by checking the Messages pane at the bottom. You should see "Commands completed successfully."

Getting a connection or certificate error?

If SSMS shows an error about encryption or certificates, check the Trust server certificate checkbox in the connection dialog. SSMS 22 defaults encryption to Mandatory, which requires certificate validation. For local development, enabling this checkbox tells SSMS to accept the self-signed certificate.

Help me fix my SSMS connection error.

Now load the CSV data into this table using Python and SQLAlchemy.

  • Return to your terminal (with the virtual environment active) and run the following Python script to load all three CSVs into stg_raw_orders:
import pandas as pd
from sqlalchemy import create_engine, text
import os
from pathlib import Path

# Connection setup
CONNECTION_STRING = (
    "mssql+pyodbc://localhost/BalticDW"
    "?driver=ODBC+Driver+18+for+SQL+Server"
    "&TrustServerCertificate=yes"
)
engine = create_engine(CONNECTION_STRING, fast_executemany=True)

# Resolve paths relative to this script, not the terminal's working directory
PROJECT_DIR = Path(__file__).resolve().parent
SOURCE_DIR = PROJECT_DIR / "source_data"

# Load all country CSVs into the flat staging table
files = ["estonia_orders.csv", "latvia_orders.csv", "lithuania_orders.csv"]

for f in files:
    filepath = os.path.join(SOURCE_DIR, f)
    if os.path.exists(filepath):
        df = pd.read_csv(filepath)
        df.to_sql("stg_raw_orders", engine, if_exists="append", index=False)
        print(f"Loaded {len(df)} rows from {f}")

What does this code do?

  • The connection string uses Windows Authentication (no username/password in the URL) and TrustServerCertificate=yes because ODBC Driver 18 defaults encryption to on.
  • The fast_executemany=True parameter enables efficient batch inserts for SQL Server.
  • Each CSV is appended to the same stg_raw_orders table, creating one big flat pile of data.
  • Save this as load_staging.py in your project folder and run it:
python load_staging.py

You should see three lines confirming 50 rows loaded from each file.

  • Switch to SSMS and verify the row count. Click New Query and run:
SELECT COUNT(*) AS total_rows FROM stg_raw_orders;

You should see 150 in the results. All three countries are now merged into one flat table.

Seeing 0 rows or a connection error in Python?

  • If you see a pyodbc error about the driver, confirm ODBC Driver 18 is installed by checking odbcad32.exe (search Windows for "ODBC Data Sources"). The driver name must match exactly: ODBC Driver 18 for SQL Server.
  • If you see 0 rows, ensure you ran generate_source_data.py first so the CSVs exist.

Help me fix my Python to SQL Server connection.

Query the flat table and spot the problems

You have 150 rows in one table. Time to ask a simple business question: how many unique customers do we have across all three countries?

  • In SSMS, open a new query window (make sure it targets BalticDW in the database dropdown).

Before you run the next query, predict: if each country has 5 customers, how many distinct customer names would you expect across all three countries?

  • Run this query to group customers by name and see how they appear across countries:
SELECT customer_name, COUNT(*) AS order_count, COUNT(DISTINCT country_code) AS countries
FROM stg_raw_orders
GROUP BY customer_name
ORDER BY countries DESC, order_count DESC;

Look at the results. You will see customers like "Baltic Fuel" appearing with orders from multiple countries, each tracked under a different customer ID. The flat table has no way to tell that these are the same economic entity.

  • Now run a revenue query to see how the flat approach inflates totals:
SELECT country_code, COUNT(*) AS orders, SUM(line_total) AS revenue
FROM stg_raw_orders
GROUP BY country_code;

What problems does this reveal?

  • Duplicate customer identities: the same real-world customer (e.g., Baltic Fuel) has separate IDs per country. Any report grouping by customer_name will double- or triple-count their activity.
  • No history tracking: if a customer changes segment (say from "Standard" to "Premium"), the flat table overwrites the old value. You lose all historical context.
  • No referential integrity: there is nothing preventing a typo in a product name from creating a phantom product in your reports.

This is the exact gap that a star schema solves. By separating descriptive attributes into dimension tables with surrogate keys, you eliminate duplicates and gain the ability to track changes over time.

You have proven the problem. A flat merge of multi-country ERP data produces unreliable numbers because it cannot deduplicate customers or track changes. Next, you will design a star schema that solves both problems.

Design and Create the Star Schema

In the previous step, you loaded all three country files into a single flat table and saw the result: duplicate customers, no change history, and inflated totals when you try to aggregate.

A star schema solves these problems by separating stable descriptive data (dimensions) from measurable business events (facts). In this step, you will create the dimension and fact tables that form your warehouse structure.

In this step, get ready to:
  • Create dimension tables for dates, products, subsidiaries, and customers.
  • Create a fact table that references all dimensions through surrogate keys.
  • Populate the date and subsidiary dimensions with reference data.
Create the dimension tables

Dimension tables hold the descriptive attributes that give context to your numbers. Each dimension gets a surrogate key (an auto-incrementing integer) that the fact table uses to look up details.

  • Switch back to schema.sql in VS Code.
  • Below the GO that closes the stg_raw_orders table, add the DimDate and DimProduct table definitions:
-- Dimension: Date
IF OBJECT_ID('DimDate', 'U') IS NOT NULL DROP TABLE DimDate;
CREATE TABLE DimDate (
    date_key INT NOT NULL PRIMARY KEY,
    full_date DATE NOT NULL,
    year INT NOT NULL,
    quarter INT NOT NULL,
    month INT NOT NULL,
    month_name NVARCHAR(20) NOT NULL,
    day_of_week INT NOT NULL,
    day_name NVARCHAR(20) NOT NULL
);
GO

-- Dimension: Product
IF OBJECT_ID('DimProduct', 'U') IS NOT NULL DROP TABLE DimProduct;
CREATE TABLE DimProduct (
    product_key INT IDENTITY(1,1) PRIMARY KEY,
    product_id NVARCHAR(10) NOT NULL,
    product_name NVARCHAR(200) NOT NULL,
    category NVARCHAR(50),
    brand NVARCHAR(50)
);
GO

What does this code do?

  • DimDate uses a date_key in YYYYMMDD integer format (e.g. 20240115). This is the standard pattern for date dimensions because integer joins are faster than date comparisons.
  • DimProduct uses IDENTITY(1,1) to generate surrogate keys automatically. The product_id column keeps the natural key from the source system.
  • The IF OBJECT_ID ... DROP TABLE pattern makes the script idempotent. You can run it repeatedly without errors.
  • Below the DimProduct block, add the DimSubsidiary and DimCustomer table definitions:
-- Dimension: Subsidiary
IF OBJECT_ID('DimSubsidiary', 'U') IS NOT NULL DROP TABLE DimSubsidiary;
CREATE TABLE DimSubsidiary (
    subsidiary_key INT IDENTITY(1,1) PRIMARY KEY,
    country_code NVARCHAR(5) NOT NULL,
    country_name NVARCHAR(50) NOT NULL,
    subsidiary_name NVARCHAR(100) NOT NULL
);
GO

-- Dimension: Customer (SCD Type 2)
IF OBJECT_ID('DimCustomer', 'U') IS NOT NULL DROP TABLE DimCustomer;
CREATE TABLE DimCustomer (
    customer_key INT IDENTITY(1,1) PRIMARY KEY,
    customer_id NVARCHAR(20) NOT NULL,
    customer_name NVARCHAR(100) NOT NULL,
    segment NVARCHAR(50),
    city NVARCHAR(100),
    country_code NVARCHAR(5),
    effective_date DATE NOT NULL,
    expiry_date DATE NOT NULL DEFAULT '9999-12-31',
    is_current BIT NOT NULL DEFAULT 1
);
GO

What is SCD Type 2?

SCD Type 2 (Slowly Changing Dimension Type 2) tracks the full history of changes to a record. When a customer's segment changes, you don't overwrite the old value. Instead you expire the old row and insert a new current row.

The three SCD columns work together: effective_date marks when a record became active, expiry_date marks when it was superseded (defaulting to 9999-12-31 for current rows), and is_current is a quick filter for the latest version.

  • Save schema.sql.
  • Switch to SSMS. Open a new query window connected to BalticDW (click New Query in the toolbar, then ensure the database dropdown shows BalticDW).
  • Paste just the four dimension CREATE TABLE statements you added (from -- Dimension: Date through the final GO after DimCustomer) and click Execute.
  • In the Object Explorer on the left, right-click Tables under BalticDW and select Refresh. You should see DimDate, DimProduct, DimSubsidiary, and DimCustomer alongside stg_raw_orders.

Tables not appearing?

  • Check the database dropdown in the query window toolbar shows BalticDW, not master. Running DDL against the wrong database creates tables there instead.
  • Look for red underlines in the Messages pane. A missing GO between statements can cause batch errors.
  • Help me debug my dimension table creation in SSMS.
Create the fact table

The fact table stores the measurable events. Each row represents one order line, and every descriptive attribute is replaced by a foreign key pointing to the relevant dimension.

  • In schema.sql, add the FactSalesOrder definition below the DimCustomer block:
-- Fact: Sales Order
IF OBJECT_ID('FactSalesOrder', 'U') IS NOT NULL DROP TABLE FactSalesOrder;
CREATE TABLE FactSalesOrder (
    fact_key INT IDENTITY(1,1) PRIMARY KEY,
    order_id NVARCHAR(20) NOT NULL,
    order_date_key INT NOT NULL,
    product_key INT NOT NULL,
    customer_key INT NOT NULL,
    subsidiary_key INT NOT NULL,
    quantity INT NOT NULL,
    unit_price DECIMAL(10,2) NOT NULL,
    line_total DECIMAL(12,2) NOT NULL,
    CONSTRAINT FK_Fact_Date FOREIGN KEY (order_date_key) REFERENCES DimDate(date_key),
    CONSTRAINT FK_Fact_Product FOREIGN KEY (product_key) REFERENCES DimProduct(product_key),
    CONSTRAINT FK_Fact_Customer FOREIGN KEY (customer_key) REFERENCES DimCustomer(customer_key),
    CONSTRAINT FK_Fact_Subsidiary FOREIGN KEY (subsidiary_key) REFERENCES DimSubsidiary(subsidiary_key)
);
GO

What does this code do?

  • The four CONSTRAINT FK_* lines enforce referential integrity. SQL Server will reject any fact row whose keys don't exist in the referenced dimension.
  • The measures (quantity, unit_price, line_total) are the numeric values you aggregate in queries. Everything else is a key that joins to a dimension for context.
  • The order_date_key is an integer matching DimDate's date_key format (YYYYMMDD), not a DATE column. This is what makes the star schema fast for date-based filtering.
  • Save schema.sql.
  • In SSMS, paste the FactSalesOrder CREATE TABLE statement into your query window and click Execute.
  • Refresh the Tables node in Object Explorer. You should now see FactSalesOrder listed alongside the four dimension tables.

Seeing a foreign key error?

  • The fact table references DimDate, DimProduct, DimCustomer, and DimSubsidiary. If any of those do not exist yet, the CREATE TABLE will fail. Run the dimension DDL first, then retry the fact table.
  • Help me fix my FactSalesOrder foreign key constraint error.
Populate the reference dimensions

Two of your dimensions contain static reference data that doesn't come from the ERP source files. DimDate needs a row for every calendar day your orders could land on. DimSubsidiary needs the three Baltic country subsidiaries.

  • In schema.sql, add the DimDate population script below the FactSalesOrder block:
-- Populate DimDate (2024-2025)
DECLARE @start DATE = '2024-01-01';
DECLARE @end DATE = '2025-12-31';
DECLARE @current DATE = @start;

WHILE @current <= @end
BEGIN
    INSERT INTO DimDate (date_key, full_date, year, quarter, month, month_name, day_of_week, day_name)
    VALUES (
        CAST(FORMAT(@current, 'yyyyMMdd') AS INT),
        @current,
        YEAR(@current),
        DATEPART(QUARTER, @current),
        MONTH(@current),
        DATENAME(MONTH, @current),
        DATEPART(WEEKDAY, @current),
        DATENAME(WEEKDAY, @current)
    );
    SET @current = DATEADD(DAY, 1, @current);
END
GO

What does this code do?

This loop generates one row per day from January 1, 2024 through December 31, 2025. That is 731 days (366 for leap year 2024 plus 365 for 2025). Each row pre-computes useful attributes like quarter and month name so reporting queries never need date math.

The FORMAT(@current, 'yyyyMMdd') expression converts a date like 2024-03-15 into the integer 20240315, which becomes the date_key used in the fact table.

  • In SSMS, paste the DimDate population script and click Execute.
  • Verify the row count by running this query:
SELECT COUNT(*) AS date_rows FROM DimDate;

You should see 731 in the result.

  • Now add the DimSubsidiary population below the DimDate loop in schema.sql:
-- Populate DimSubsidiary
INSERT INTO DimSubsidiary (country_code, country_name, subsidiary_name) VALUES
('EE', 'Estonia', 'Jungent Estonia OÜ'),
('LV', 'Latvia', 'Jungent Latvia SIA'),
('LT', 'Lithuania', 'Jungent Lithuania UAB');
GO
  • Paste and execute this INSERT in SSMS.
  • Verify by running:
SELECT * FROM DimSubsidiary;

You should see 3 rows: Estonia, Latvia, and Lithuania with their subsidiary names.

  • Save schema.sql in VS Code. Your schema file now contains the complete warehouse DDL.

DimDate shows 0 rows?

  • If you ran the full schema.sql file (which includes the DROP TABLE for DimDate) after populating it, the data was wiped. Run just the population WHILE loop again without re-running the DROP/CREATE.
  • Check the Messages pane for any errors during the WHILE loop execution. A typo in the FORMAT string will cause a silent failure.
  • Help me figure out why my DimDate population returned 0 rows.

Before you do anything else, predict what you expect to see if you expand the BalticDW tables list and check row counts across all tables.

  • Run this final verification query in SSMS:
SELECT 'stg_raw_orders' AS table_name, COUNT(*) AS row_count FROM stg_raw_orders
UNION ALL
SELECT 'DimDate', COUNT(*) FROM DimDate
UNION ALL
SELECT 'DimProduct', COUNT(*) FROM DimProduct
UNION ALL
SELECT 'DimSubsidiary', COUNT(*) FROM DimSubsidiary
UNION ALL
SELECT 'DimCustomer', COUNT(*) FROM DimCustomer
UNION ALL
SELECT 'FactSalesOrder', COUNT(*) FROM FactSalesOrder;

You should see: stg_raw_orders with 150 rows, DimDate with 731, DimSubsidiary with 3, and DimProduct, DimCustomer, and FactSalesOrder all with 0. The empty tables will be populated by your ETL pipeline in the next step.

✔️ Awesome, I've got everything!

Great. Double check you have saved schema.sql with all the dimension tables, the fact table, and the population scripts.

ⓧ I'd like to double check the full code

Here is the complete schema.sql file. Scroll up and confirm each section is present in this order:

  • CREATE DATABASE BalticDW (with IF NOT EXISTS check).
  • CREATE TABLE stg_raw_orders (the flat staging table).
  • CREATE TABLE DimDate (8 columns, integer primary key).
  • CREATE TABLE DimProduct (surrogate key + 4 columns).
  • CREATE TABLE DimSubsidiary (surrogate key + 3 columns).
  • CREATE TABLE DimCustomer (SCD Type 2 with effective_date, expiry_date, is_current).
  • CREATE TABLE FactSalesOrder (4 FK constraints referencing all dimensions).
  • WHILE loop populating DimDate with 731 rows (2024-01-01 to 2025-12-31).
  • INSERT populating DimSubsidiary with 3 Baltic subsidiaries.

Your star schema is in place. The structure separates facts from dimensions, and the SCD Type 2 columns on DimCustomer are ready to track changes over time. Next up, you will build the Python ETL pipeline that transforms raw staging data into properly dimensioned records.

Build the ETL Pipeline with SCD Type 2

Your star schema is built and waiting for data. The dimension tables have their columns and constraints defined, but they sit empty.

In this step, you will write the Python pipeline that reads from the flat staging table and populates each dimension and fact table correctly. The hardest part is the customer dimension: it needs to detect whether a customer already exists and whether their attributes have changed, then preserve history using SCD Type 2 logic.

In this step, get ready to:
  • Build dimension loading functions that deduplicate products and track customer changes.
  • Build the fact loading function that resolves surrogate keys from each dimension.
  • Run the full ETL pipeline end-to-end and verify all 150 fact rows load correctly.
Create the pipeline connection and staging loader

The ETL pipeline lives in a single Python file that connects to SQL Server using SQLAlchemy and pyodbc. It uses Windows Authentication so no credentials are needed in the connection string.

  • In VS Code, create a new file called etl_pipeline.py in your project folder.
  • Add the imports and connection setup by pasting this code:
import pandas as pd
from sqlalchemy import create_engine, text
from datetime import date
import os
from pathlib import Path

# Connection setup
CONNECTION_STRING = (
    "mssql+pyodbc://localhost/BalticDW"
    "?driver=ODBC+Driver+18+for+SQL+Server"
    "&TrustServerCertificate=yes"
)
engine = create_engine(CONNECTION_STRING, fast_executemany=True)

# Resolve paths relative to this script, not the terminal's working directory
PROJECT_DIR = Path(__file__).resolve().parent
SOURCE_DIR = PROJECT_DIR / "source_data"

What does this code do?

The connection string tells SQLAlchemy to use the mssql+pyodbc dialect, connect to localhost, and target the BalticDW database you created in Step 3.

The TrustServerCertificate=yes flag is needed because ODBC Driver 18 defaults encryption to on, and your local SQL Server uses a self-signed certificate. The fast_executemany=True option enables efficient bulk inserts.

Path(__file__).resolve().parent anchors file paths to this script's directory, just like in generate_source_data.py. This ensures the pipeline finds the CSV files regardless of your terminal's working directory.

  • Below the connection setup, add the staging loader function:
def load_staging():
    """Load all country CSVs into the flat staging table."""
    files = [
        "estonia_orders.csv",
        "latvia_orders.csv",
        "lithuania_orders.csv",
    ]

    with engine.begin() as conn:
        conn.execute(text("TRUNCATE TABLE stg_raw_orders"))

    for f in files:
        filepath = os.path.join(SOURCE_DIR, f)
        if os.path.exists(filepath):
            df = pd.read_csv(filepath)
            df.to_sql("stg_raw_orders", engine, if_exists="append", index=False)
            print(f"Loaded {len(df)} rows from {f}")

What does this code do?

The load_staging() function truncates the staging table first (so re-runs are safe), then reads each country CSV and appends it using pandas to_sql. This gives you a clean reload every time the pipeline runs.

  • Save etl_pipeline.py.
  • Test the connection and staging loader by running this command in your terminal:
python -c "from etl_pipeline import load_staging; load_staging()"

You should see three lines confirming 50 rows loaded from each CSV file.

Seeing a connection error?

  • Make sure the SQL Server service is running. Open services.msc and confirm the SQL Server service shows Running.
  • Double check the driver name in your connection string matches exactly: ODBC Driver 18 for SQL Server.
  • Make sure your virtual environment is activated (you should see (venv) in your terminal prompt).

Why can't my Python script connect to SQL Server?

Build the dimension loading functions

Dimension loaders read distinct values from the flat staging table and insert them into the proper dimension tables. The product loader is straightforward. The customer loader is where the SCD Type 2 logic lives.

  • In etl_pipeline.py, add the product dimension loader below load_staging():
def load_dim_product():
    """Load distinct products from staging into DimProduct."""
    query = """
        SELECT DISTINCT product_id, product_name, product_category, product_brand
        FROM stg_raw_orders
    """
    with engine.begin() as conn:
        staging_products = pd.read_sql(query, conn)

        existing = pd.read_sql(
            text("SELECT product_id FROM DimProduct"), conn
        )
        existing_ids = set(existing["product_id"].tolist())

        new_products = staging_products[
            ~staging_products["product_id"].isin(existing_ids)
        ]

        if not new_products.empty:
            new_products = new_products.rename(columns={
                "product_category": "category",
                "product_brand": "brand",
            })
            new_products[["product_id", "product_name", "category", "brand"]].to_sql(
                "DimProduct", conn, if_exists="append", index=False
            )
            print(f"Inserted {len(new_products)} new products")
        else:
            print("No new products to insert")

What does this code do?

The function queries distinct products from staging, compares them against what already exists in DimProduct, and only inserts genuinely new products. This makes the loader idempotent. Running it twice produces the same result.

The column rename from product_category to category matches the staging column names to the dimension table schema.

  • Now start the customer dimension loader. Add the function setup and loop skeleton below load_dim_product():
def load_dim_customer():
    """Load customers with SCD Type 2 logic."""
    query = """
        SELECT DISTINCT customer_id, customer_name, customer_segment,
                        customer_city, country_code
        FROM stg_raw_orders
    """
    with engine.begin() as conn:
        staging_customers = pd.read_sql(query, conn)

        existing = pd.read_sql(
            text("SELECT customer_key, customer_id, customer_name, segment, city, country_code FROM DimCustomer WHERE is_current = 1"),
            conn,
        )

        today = date.today().isoformat()

        for _, row in staging_customers.iterrows():
            match = existing[existing["customer_id"] == row["customer_id"]]

What does this code do?

The function fetches all distinct customers from staging, then loads the current set of customers from DimCustomer (only rows where is_current = 1). For each staging customer, it looks up a match by natural key (customer_id).

The today variable sets the effective date for any new or updated dimension rows.

  • Save the file, then test that staging and the product dimension load correctly by running:
python -c "from etl_pipeline import load_staging, load_dim_product; load_staging(); load_dim_product()"

You should see the three staging load messages followed by Inserted 8 new products.

Seeing 0 new products?

If you ran the pipeline earlier and products already exist in DimProduct, the idempotent check skips them. Run TRUNCATE TABLE DimProduct in SSMS first, then re-run.

Why does my product dimension loader say no new products to insert?

  • Back in etl_pipeline.py, continue the load_dim_customer function. Add the if/else logic inside the for loop, directly below the match line:
            if match.empty:
                conn.execute(
                    text("""
                        INSERT INTO DimCustomer
                            (customer_id, customer_name, segment, city, country_code, effective_date, expiry_date, is_current)
                        VALUES
                            (:cid, :cname, :seg, :city, :cc, :eff, '9999-12-31', 1)
                    """),
                    {
                        "cid": row["customer_id"],
                        "cname": row["customer_name"],
                        "seg": row["customer_segment"],
                        "city": row["customer_city"],
                        "cc": row["country_code"],
                        "eff": today,
                    },
                )
            else:
                existing_row = match.iloc[0]
                changed = (
                    existing_row["segment"] != row["customer_segment"]
                    or existing_row["city"] != row["customer_city"]
                    or existing_row["customer_name"] != row["customer_name"]
                )

What does this code do?

If no matching customer exists (match.empty), the customer is new. It gets inserted with is_current=1 and an expiry date of 9999-12-31 (meaning it has no end date yet).

If the customer does exist, the code compares segment, city, and name against the current dimension row. The changed boolean captures whether any attribute differs. The next chunk handles what to do when a change is detected.

  • Still inside the else block, add the SCD Type 2 update logic below the changed variable. Then add the closing print statement at the function level:
                if changed:
                    conn.execute(
                        text("""
                            UPDATE DimCustomer
                            SET expiry_date = :exp, is_current = 0
                            WHERE customer_key = :key
                        """),
                        {"exp": today, "key": int(existing_row["customer_key"])},
                    )
                    conn.execute(
                        text("""
                            INSERT INTO DimCustomer
                                (customer_id, customer_name, segment, city, country_code, effective_date, expiry_date, is_current)
                            VALUES
                                (:cid, :cname, :seg, :city, :cc, :eff, '9999-12-31', 1)
                        """),
                        {
                            "cid": row["customer_id"],
                            "cname": row["customer_name"],
                            "seg": row["customer_segment"],
                            "city": row["customer_city"],
                            "cc": row["country_code"],
                            "eff": today,
                        },
                    )
                    print(f"SCD2 update: {row['customer_id']} segment changed")

        print("Customer dimension loaded")

What is SCD Type 2 doing here?

When a customer's attributes have changed, two things happen. First, the old row gets expired: is_current is set to 0 and expiry_date is set to today. Second, a brand new row is inserted with the updated attributes and is_current=1.

This means historical fact rows still point to the old surrogate key (the expired row), preserving what the customer looked like when those orders were placed. New facts reference the new surrogate key with the updated attributes.

  • Save the file and test all dimension loaders by running:
python -c "from etl_pipeline import load_staging, load_dim_product, load_dim_customer; load_staging(); load_dim_product(); load_dim_customer()"

You should see the staging messages, the product insert, and then Customer dimension loaded. Since this is the first run, all 15 customers are new inserts. No SCD2 update messages appear yet.

Build the fact loading function

The fact loader is the final piece. It joins staging rows against all four dimensions to resolve surrogate keys, then inserts the results into FactSalesOrder.

  • In etl_pipeline.py, add the fact loading function below load_dim_customer():
def load_fact_sales():
    """Load facts by resolving surrogate keys from dimensions."""
    query = """
        SELECT
            s.order_id,
            CAST(FORMAT(s.order_date, 'yyyyMMdd') AS INT) AS order_date_key,
            p.product_key,
            c.customer_key,
            sub.subsidiary_key,
            s.quantity,
            s.unit_price,
            s.line_total
        FROM stg_raw_orders s
        INNER JOIN DimProduct p ON s.product_id = p.product_id
        INNER JOIN DimCustomer c ON s.customer_id = c.customer_id AND c.is_current = 1
        INNER JOIN DimSubsidiary sub ON s.country_code = sub.country_code
        WHERE s.order_id NOT IN (SELECT order_id FROM FactSalesOrder)
    """
    with engine.begin() as conn:
        facts = pd.read_sql(query, conn)

        if not facts.empty:
            facts.to_sql("FactSalesOrder", conn, if_exists="append", index=False)
            print(f"Inserted {len(facts)} fact rows")
        else:
            print("No new facts to insert")

What does this code do?

The SQL query joins stg_raw_orders to each dimension table, resolving natural keys (like product_id) into surrogate keys (like product_key). The date key is derived by formatting the order date as YYYYMMDD integer.

The WHERE s.order_id NOT IN (SELECT order_id FROM FactSalesOrder) clause ensures idempotency. Orders already loaded are skipped on re-runs.

  • Add the main execution block at the bottom of the file:
if __name__ == "__main__":
    print("=== Baltic DW ETL Pipeline ===")
    print("\n1. Loading staging...")
    load_staging()
    print("\n2. Loading DimProduct...")
    load_dim_product()
    print("\n3. Loading DimCustomer (SCD Type 2)...")
    load_dim_customer()
    print("\n4. Loading FactSalesOrder...")
    load_fact_sales()
    print("\n=== ETL Complete ===")

What does this code do?

The if __name__ == "__main__" guard means this block only runs when you execute the file directly (python etl_pipeline.py). It orchestrates the full pipeline in the correct order: staging first, then dimensions, then facts.

  • Save etl_pipeline.py.
Run the full ETL pipeline end-to-end

All four functions are ready. Time to run the complete pipeline and verify the warehouse is populated.

  • Before you run this, predict: how many fact rows do you expect to see inserted? Think about the number of staging rows and whether any would be skipped.
  • Run the full pipeline:
python etl_pipeline.py

You should see output showing all stages complete, ending with Inserted 150 fact rows and === ETL Complete ===.

Why 150 fact rows?

Each of the 3 country CSVs contains 50 order lines. All 150 staging rows have valid dimension matches (every product, customer, subsidiary, and date exists in the corresponding dimension table), so all 150 resolve to fact rows.

  • Switch to SSMS and verify the fact table by running:
SELECT COUNT(*) FROM FactSalesOrder;

You should see 150.

  • Verify the customer dimension by running:
SELECT COUNT(*) FROM DimCustomer WHERE is_current = 1;

You should see 15. That is 5 customers per country, each with is_current = 1 and no expired rows yet.

Seeing fewer than 150 fact rows?

  • If load_fact_sales reported fewer inserts, some staging rows may lack a dimension match. Run the fact query manually in SSMS to check which joins fail.
  • If you ran the pipeline multiple times, the idempotent check skips already-loaded orders. The total in FactSalesOrder should still be 150.
  • Confirm DimDate has 731 rows and DimSubsidiary has 3 rows. Missing dimension rows cause JOIN failures.

Why does my fact table have fewer than 150 rows after running the ETL pipeline?

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py and that your SSMS queries confirmed 150 fact rows and 15 current customers.

ⓧ I'd like to double check the full code

Your etl_pipeline.py should contain these sections in order. Scroll through the file and confirm each is present:

  • Imports (pandas, sqlalchemy, datetime, os) and CONNECTION_STRING with engine setup.
  • The load_staging() function that truncates and reloads all three CSVs.
  • The load_dim_product() function that inserts distinct products with an idempotent check.
  • The load_dim_customer() function with the for loop, new-customer insert, and the else branch containing SCD Type 2 expire-and-insert logic.
  • The load_fact_sales() function that resolves surrogate keys via JOINs and inserts into FactSalesOrder.
  • The if __name__ == "__main__" block calling all four functions in order.

Your warehouse is fully loaded. Every staging row now lives as a properly keyed fact, linked to deduplicated dimensions. Next up, you will query the star schema and prove it produces correct numbers where the flat table failed.

Query the Star Schema and Compare

Your ETL pipeline loaded clean, deduplicated data into the star schema. The real test is whether this architecture produces trusted numbers that the flat staging table cannot.

In this step, you'll run the same business question against both the star schema and the flat table to see the structural difference firsthand. You'll also confirm that your pipeline's incremental load logic prevents duplicate processing on re-runs.

In this step, get ready to:
  • Write cross-border revenue queries against both the star schema and flat table to compare results.
  • Write a customer segment profitability query using proper dimension joins.
  • Verify the pipeline's incremental load logic prevents duplicate processing.
Run the cross-border revenue query against the star schema

The whole point of building the star schema was to answer business questions cleanly. The first question every FMCG distributor asks is: how much revenue did each country generate, broken down by month?

  • Switch back to SSMS and open a new query window against BalticDW.
  • Run this cross-border revenue query that joins the fact table to the subsidiary and date dimensions:
-- Cross-border revenue: star schema version
SELECT
    sub.country_name,
    d.year,
    d.month_name,
    COUNT(*) AS order_lines,
    SUM(f.line_total) AS total_revenue
FROM FactSalesOrder f
INNER JOIN DimSubsidiary sub ON f.subsidiary_key = sub.subsidiary_key
INNER JOIN DimDate d ON f.order_date_key = d.date_key
GROUP BY sub.country_name, d.year, d.month_name, d.month
ORDER BY sub.country_name, d.year, d.month;

What does this query do?

The query joins FactSalesOrder to DimSubsidiary (to get human-readable country names) and DimDate (to get year/month breakdowns). Each join uses the surrogate key your ETL pipeline resolved.

This is the star schema pattern: the fact table sits at the center and dimensions radiate outward, each adding descriptive context to the raw measures.

You should see results grouped by Estonia, Latvia, and Lithuania with monthly revenue totals. Each row represents one country-month combination with clean, deduplicated numbers.

Before you run the next query, predict: will the flat staging table produce the same totals, or different ones?

  • Now run the equivalent query against the flat staging table:
-- Cross-border revenue: flat staging table version
SELECT
    country_code,
    YEAR(order_date) AS year,
    DATENAME(MONTH, order_date) AS month_name,
    COUNT(*) AS order_lines,
    SUM(line_total) AS total_revenue
FROM stg_raw_orders
GROUP BY country_code, YEAR(order_date), DATENAME(MONTH, order_date), MONTH(order_date)
ORDER BY country_code, YEAR(order_date), MONTH(order_date);

Why are the totals the same here?

For a simple SUM by country and month, both tables return the same totals because the underlying 150 rows are identical. The flat table gets lucky on this query.

The difference shows up when you ask more complex questions. The flat table has no protection against double-loading, no way to track customer changes, and no referential integrity. Run the next query to see where it breaks down.

  • Run this query to count distinct customers per country from the flat table:
-- Customer count comparison: flat table
SELECT
    country_code,
    COUNT(DISTINCT customer_id) AS unique_customers,
    COUNT(DISTINCT customer_name) AS unique_names,
    COUNT(*) AS total_rows_per_country
FROM stg_raw_orders
GROUP BY country_code;

What's wrong with the flat table here?

The flat table shows 5 customers per country, but it cannot tell you that Baltic Fuel operates across all three countries as the same business entity. Each row repeats customer attributes (name, segment, city) redundantly.

If Baltic Fuel's segment changes from Standard to Premium, some order rows show the old segment and others show the new one. There is no single source of truth. The star schema's DimCustomer with SCD Type 2 solves this by maintaining a versioned history per customer.

🙋‍♀️ Query returning zero rows?

Make sure your query window is connected to BalticDW (check the database dropdown in the SSMS toolbar). If it says master, switch it to BalticDW.

help me debug my star schema query in SSMS

Write a customer segment profitability query

The second business question combines customer segments with product categories to show where the money actually comes from across all three Baltic countries.

  • Run this customer segment profitability query in SSMS:
-- Customer segment profitability by product category
SELECT
    c.segment,
    p.category,
    COUNT(*) AS order_lines,
    SUM(f.quantity) AS total_units,
    SUM(f.line_total) AS total_revenue
FROM FactSalesOrder f
INNER JOIN DimCustomer c ON f.customer_key = c.customer_key
    AND c.is_current = 1
INNER JOIN DimProduct p ON f.product_key = p.product_key
GROUP BY c.segment, p.category
ORDER BY total_revenue DESC;

What does this query do?

  • Joins FactSalesOrder to DimCustomer (filtering is_current = 1 so you only see the latest version of each customer).
  • Joins to DimProduct to break revenue down by product category (Oral Care, Lubricants, Tissue, etc.).
  • Groups by segment and category so you see which customer types buy which product types.

You should see results broken down by segments like Retail, HORECA, and Standard, each with revenue per product category. This query is impossible to write reliably against the flat table because the flat table has no enforced relationship between customers and their current attributes.

Why filter on is_current = 1?

DimCustomer uses SCD Type 2, which means a customer can have multiple rows (one per historical version). Filtering is_current = 1 gives you only the latest state of each customer for current reporting. For historical analysis, you would remove this filter and use date ranges instead.

Verify incremental load logic

A production ETL pipeline must handle re-runs without duplicating data. Your pipeline already has this built in. The key is the WHERE clause in load_fact_sales().

  • In VS Code, open etl_pipeline.py and find the load_fact_sales function. Locate this line inside the SQL query:
WHERE s.order_id NOT IN (SELECT order_id FROM FactSalesOrder)

What does this watermark logic do?

This clause acts as an incremental load guard. It checks each staging row's order_id against what already exists in FactSalesOrder. Only new orders (not yet loaded) get inserted.

Without this, re-running the pipeline would insert 150 duplicate rows every time. In production, this pattern is called watermark-based extraction. The watermark here is the set of already-loaded order IDs.

Before you re-run the pipeline, predict: how many new fact rows will be inserted?

  • Prove the incremental logic works by re-running the full pipeline in your terminal:
python etl_pipeline.py

You should see the pipeline print No new facts to insert at step 4. The staging data is identical to what was already loaded, so the NOT IN clause filters out all 150 rows.

  • Confirm the fact count is unchanged by running this in SSMS:
SELECT COUNT(*) AS total_facts FROM FactSalesOrder;

You should still see 150 rows. The pipeline re-ran without creating duplicates.

Seeing more than 150 rows?

If your count is higher than 150, the NOT IN clause may not be matching correctly. Check that the order_id column in FactSalesOrder matches the format in stg_raw_orders (e.g., EE-ORD-10000). You can reset by truncating the fact table and re-running once.

help me debug duplicate facts in my FactSalesOrder table

💡 Why this approach over a date-based watermark?

A date-based watermark (only load orders newer than the last loaded date) is common in production. The NOT IN approach used here is simpler for a learning project because it handles out-of-order arrivals and backdated orders without extra logic.

In a high-volume production system, you would combine both: a date watermark for performance (avoids scanning all historical IDs) plus a deduplication check as a safety net.

You now have a fully functional star schema warehouse that produces trusted cross-border reports and handles re-runs safely. The flat staging table gave the same simple totals but offered no protection against duplicates, no change tracking, and no clean dimension joins for complex analysis.

Next up, you'll put the SCD Type 2 logic to the real test by processing a batch of changed customer data and verifying that history is preserved correctly.

Secret mission

Process a Change Batch and Verify History

In the real world, customers change segments, addresses, and contact details constantly. Can your pipeline handle a second batch where three customers got promoted from Standard to Premium, while still preserving what segment they were in when historical orders were placed?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally on your machine, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • SQL Server 2025 Developer Edition (local install, free, no ongoing cost).
  • BalticDW database with star schema tables and ETL data.
  • Python virtual environment with pyodbc, sqlalchemy, and pandas packages.
  • Project folder containing Python scripts and source CSV files.

Keep everything running

No action needed. Choose this if you plan to continue building on this warehouse or move into the next project (adding a dbt reporting layer and Power BI dashboards).

  • SQL Server runs as a Windows service in the background. It uses minimal CPU and memory when idle.
  • Your BalticDW database, Python scripts, and source data files all remain on disk ready to use.

Pause - I'll come back to this later

Stop the SQL Server service to free up memory while keeping all your files and database intact. You can restart it any time you return.

  • Press the Windows key, type services.msc, and press Enter to open the Services panel.
  • Scroll down and find SQL Server (MSSQLSERVER) in the list.
  • Right-click it and select Stop.
  • When you come back, right-click the same service and select Start to resume.

Delete - I don't want to use this again

Remove all project resources and reclaim disk space. This removes your database, scripts, and the SQL Server installation entirely.

  • Delete your project folder (which contains the virtual environment, Python scripts, and source data) by running this in your terminal:
cd ~\Desktop
rmdir /s /q baltic-dw
  • Uninstall SQL Server 2025 by opening Settings, selecting Apps, then Installed apps. Search for SQL Server 2025 and click Uninstall. This frees approximately 2 GB of disk space and removes the BalticDW database.
  • Uninstall SSMS 22 from the same Installed apps list. Search for SQL Server Management Studio and click Uninstall.
  • Uninstall ODBC Driver 18 from the same list. Search for ODBC Driver 18 for SQL Server and click Uninstall.

Nice Work!

Nice Work!

You just built a production-grade star schema data warehouse from raw multi-country ERP data. You can now explain and demonstrate the full pipeline to an interviewer: from messy flat exports, through a Python ETL pipeline with change detection, into clean dimensional tables that produce trusted cross-border reporting.

You've learned how to:

  • Design and build a star schema data warehouse on SQL Server 2025 with dimension tables (DimDate, DimProduct, DimCustomer, DimSubsidiary) and a fact table (FactSalesOrder) that produce correct, deduplicated cross-border queries.
  • Write a Python ETL pipeline using SQLAlchemy and pyodbc that consolidates multi-country ERP data into a unified warehouse with surrogate key resolution and incremental watermark-based extraction.
  • Implement SCD Type 2 change tracking on the customer dimension with effective_date, expiry_date, and is_current columns that preserve full attribute history for temporal reporting.
  • Secret Mission: Processed a change batch proving SCD Type 2 correctly preserves history while tracking current state, with historical facts pointing to old dimension rows and new facts pointing to updated rows.

Ready to quiz yourself?