Insurance Claims Data Warehouse

Build a Python ETL pipeline loading insurance data into a PostgreSQL star schema.

Introduction

30 Second Summary

Every time someone files an insurance claim, their data joins thousands of other records that a company needs to organize, query, and analyze. Spotting fraud, assessing risk, and making fast decisions all depend on data that is clean and structured for analytics.

In this project, you will build a Python ETL pipeline that takes raw insurance claims data, cleans it, and loads it into a properly modeled PostgreSQL data warehouse. You will design a star schema and run analytical queries that answer real business questions.

What You'll Build

A complete ETL pipeline that extracts 1,000 raw insurance claims records, transforms them with pandas, and loads them into a star schema in PostgreSQL with four dimension tables and one fact table.

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

  • A Python ETL script that extracts, cleans, and loads insurance claims data into PostgreSQL.
  • A star schema data warehouse with four dimension tables and one fact table linked by foreign keys.
  • Analytical SQL queries that reveal fraud rates, claim severity patterns, and high-risk customer segments.
  • Secret Mission: Add production-grade data quality validation checks to make your pipeline interview-ready.

Are there any prerequisites?

This project assumes intermediate Python knowledge and basic SQL familiarity. You will also need PostgreSQL installed on your machine before starting.

Set Up Your Project Environment

Before you can build an ETL pipeline, you need a clean workspace with the right tools installed.

In this step, you will create a project folder, set up a Python virtual environment with your dependencies, create a PostgreSQL database to hold the warehouse, and prepare the file structure for your code.

In this step, get ready to:
  • Create a project directory and install Python dependencies.
  • Create the PostgreSQL data warehouse database.
  • Set up the file structure and download the source data.
Create your project directory and install dependencies

A virtual environment keeps your project's packages isolated from other Python projects on your machine. This prevents version conflicts between different projects.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • Open a terminal inside VS Code by pressing Ctrl+` (backtick key, top-left of your keyboard).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir insurance-claims-dw
cd insurance-claims-dw
  • Confirm you are in the right directory by running:
pwd

You should see a path ending in insurance-claims-dw, like /Users/yourname/Desktop/insurance-claims-dw.

  • Create a Python virtual environment by running:

macOS

python3 -m venv venv
  • Activate the virtual environment by running:
source venv/bin/activate

Windows

python -m venv venv
  • Activate the virtual environment by running:
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell, then try the activate command again.

On macOS, make sure you typed source before the path. Without it, the shell does not run the activation script.

Help me activate my Python virtual environment.

  • Install the project dependencies by running:
pip install pandas==2.3.3 sqlalchemy==2.0.50 psycopg2-binary==2.9.12

What are these packages?

  • pandas is a data manipulation library. You will use it to read, clean, and reshape the raw CSV data.
  • sqlalchemy is a database toolkit that lets Python talk to PostgreSQL using a connection engine.
  • psycopg2-binary is the PostgreSQL adapter that SQLAlchemy uses under the hood to send queries to your database.
  • Verify the packages installed correctly by running:
pip list

You should see pandas, SQLAlchemy, and psycopg2-binary listed with their version numbers.

Seeing an installation error?

Make sure (venv) appears in your terminal prompt before running pip install. If it does not, re-run the activation command from the previous step.

If you see a permissions error, you may have accidentally used your system Python instead of the virtual environment.

Help me fix my pip install error.

Create the PostgreSQL database

Your data warehouse needs a dedicated database in PostgreSQL. This keeps your warehouse tables separate from any other data on your local server.

  • Connect to your local PostgreSQL server by running:

macOS

psql postgres

Windows

psql -U postgres

Why the -U flag?

On Windows, PostgreSQL typically requires you to specify the username explicitly. The -U postgres flag connects as the default superuser. You may be prompted for the password you set during PostgreSQL installation.

You should see the postgres=# prompt, which means you are inside the PostgreSQL interactive terminal.

  • Create the data warehouse database by running this SQL command:
CREATE DATABASE insurance_dw;
  • Verify the database was created by listing all databases:
\l

You should see insurance_dw in the list of databases.

  • Exit the PostgreSQL terminal by running:
\q

Can't connect to PostgreSQL?

Make sure the PostgreSQL service is running. On macOS, check with brew services list if you installed via Homebrew. On Windows, search for Services in the Start menu and confirm the PostgreSQL service shows as Running.

Help me connect to my local PostgreSQL server.

Create the project files and download the source data

Your project needs two code files and a directory to hold the raw source data.

  • Create the project files and data directory by running these commands:
touch etl_pipeline.py schema.sql
mkdir data

What are these files for?

  • etl_pipeline.py will contain your Python ETL code that extracts, transforms, and loads the data.
  • schema.sql will define the star schema tables in your PostgreSQL database.
  • data/ will hold the raw CSV file you are about to download.
  • Download the insurance claims dataset into your data/ folder by running:
curl -o data/insurance_claims.csv https://raw.githubusercontent.com/mwitiderrick/insurancedata/master/insurance_claims.csv

What is this dataset?

This CSV contains 1,000 rows of auto insurance claims data with 38 columns. It covers policy details, incident information, demographics, vehicle data, and claim amounts.

The data has realistic messiness that you will clean in later steps. Some columns use ? as a placeholder for missing values instead of leaving them blank.

  • Confirm all files were created by running:
ls

You should see data, etl_pipeline.py, schema.sql, and venv in the output.

  • Verify the CSV downloaded correctly by checking its size:
wc -l data/insurance_claims.csv

You should see 1001 lines (1,000 data rows plus 1 header row).

Seeing 0 lines or a download error?

Make sure you have an active internet connection. The curl command downloads the file from GitHub.

If curl is not available on your system, try opening the URL directly in your browser and saving the file manually to your data/ folder as insurance_claims.csv.

Help me download the insurance claims CSV file.

  • Open the project folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Navigate to Desktop and select the insurance-claims-dw folder, then click Open.

You should now see your project files listed in the VS Code Explorer sidebar on the left.

Your environment is ready. Next up, you will extract the raw data and profile it to understand what needs cleaning.

Extract and Profile the Raw Data

Your environment is set up and the raw CSV is sitting in your data/ folder. Before you can transform or load anything, you need to understand what you're working with.

The first phase of any ETL pipeline is extraction. In this step, you'll write a Python function that reads the insurance claims CSV into a pandas DataFrame and profiles the data for quality issues that will need fixing later.

In this step, get ready to:
  • Write an extraction function that reads the CSV and prints profiling statistics.
  • Run the function and examine the data shape, types, and quality issues.
  • Document findings about missing value patterns and data types that need cleaning.
Add imports and configuration

Every ETL script starts with imports and configuration constants. Defining the database URL and file path as constants at the top makes the script easy to reconfigure without hunting through function bodies.

  • Open etl_pipeline.py in VS Code and add the module docstring, imports, and configuration by pasting this code:
"""Insurance Claims ETL Pipeline - Extract, Transform, Load into PostgreSQL Star Schema."""

import pandas as pd
from sqlalchemy import create_engine, text

# --- Configuration ---
DATABASE_URL = "postgresql+psycopg2://localhost/insurance_dw"
SOURCE_CSV = "data/insurance_claims.csv"

What does this code do?

  • pandas is the data manipulation library you'll use to read, clean, and reshape the CSV data.
  • sqlalchemy provides the database connection engine. You import create_engine and text now so they're ready for later steps.
  • DATABASE_URL tells SQLAlchemy how to connect to your local PostgreSQL database insurance_dw.
  • SOURCE_CSV points to the raw data file you downloaded in Step 1.
  • Save etl_pipeline.py.
  • Confirm the file has no syntax errors by running:
python etl_pipeline.py
  • You should see no output and no errors. The file only defines imports and constants so far.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • If it's not activated, run source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows) and try again.

Why can't Python find my pandas import?

Write the extraction function

The extract function reads the raw CSV and immediately prints profiling statistics. This gives you a snapshot of the data's shape, types, and quality issues in one call.

  • In etl_pipeline.py, add the extract function below the configuration constants:
# --- Extract ---
def extract(filepath):
    """Extract raw data from CSV file."""
    df = pd.read_csv(filepath)
    print(f"Extracted: {df.shape[0]} rows, {df.shape[1]} columns")
    print(f"\nColumn types:\n{df.dtypes}")
    print(f"\nNull counts:\n{df.isnull().sum()[df.isnull().sum() > 0]}")
    print(f"\nPlaceholder '?' counts:")
    question_marks = (df == "?").sum()
    print(question_marks[question_marks > 0])
    return df

What does this function do?

  • pd.read_csv(filepath) loads the entire CSV into a DataFrame. Pandas infers column types automatically.
  • df.shape reports the row and column count so you can confirm you loaded the full dataset.
  • df.isnull().sum() counts standard null values per column. This catches Python-recognized NaN entries.
  • (df == "?").sum() catches a non-standard pattern: this dataset uses the literal string ? instead of leaving cells empty. Pandas won't flag these as null unless you tell it to.
  • Save etl_pipeline.py. You won't see output yet because the function is defined but not called.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is what your complete etl_pipeline.py should look like at this point:

"""Insurance Claims ETL Pipeline - Extract, Transform, Load into PostgreSQL Star Schema."""

import pandas as pd
from sqlalchemy import create_engine, text

# --- Configuration ---
DATABASE_URL = "postgresql+psycopg2://localhost/insurance_dw"
SOURCE_CSV = "data/insurance_claims.csv"


# --- Extract ---
def extract(filepath):
    """Extract raw data from CSV file."""
    df = pd.read_csv(filepath)
    print(f"Extracted: {df.shape[0]} rows, {df.shape[1]} columns")
    print(f"\nColumn types:\n{df.dtypes}")
    print(f"\nNull counts:\n{df.isnull().sum()[df.isnull().sum() > 0]}")
    print(f"\nPlaceholder '?' counts:")
    question_marks = (df == "?").sum()
    print(question_marks[question_marks > 0])
    return df
Run the extraction and profile the data

Now you'll call the extract function to see what the raw data looks like. This is your first look at the quality issues you'll need to fix in the transform step.

  • Run the extract function against your CSV by executing this command:
python -c "from etl_pipeline import extract, SOURCE_CSV; extract(SOURCE_CSV)"
  • You should see output starting with Extracted: 1000 rows, 38 columns followed by column types, null counts, and a list of columns containing ? placeholders.

What should I see in the output?

The output reveals several data quality issues you'll address in the next step:

  • 1000 rows and 38 columns confirms the full dataset loaded.
  • The column types section shows that date columns (policy_bind_date, incident_date) are listed as object instead of datetime. They'll need parsing.
  • The ? placeholder section shows columns like collision_type, property_damage, and police_report_available that use ? instead of proper NaN for missing data.
  • The fraud_reported column uses Y and N strings instead of boolean True/False values.

Seeing a FileNotFoundError?

  • Make sure you're running the command from your project's root directory (the folder containing etl_pipeline.py and the data/ folder).
  • Check that data/insurance_claims.csv exists. Run ls data/ to verify the file is there.

Why is my extract function failing with a file error?

Why profile before transforming?

Data profiling is the detective work of data engineering. You need to discover what's wrong before writing transformation logic.

This dataset has a sneaky problem: it uses ? as a missing value marker instead of leaving cells blank. Pandas doesn't recognize ? as a null value by default, so without profiling you'd never know those values were missing.

You've profiled the raw data and identified three key issues: ? placeholder values masking missing data, date columns stored as strings, and categorical columns like fraud_reported using text instead of proper types. Next up, you'll write the transform logic to fix all of these.

Transform the Data

In the previous step, you profiled the raw data and discovered its quirks. The dataset uses ? as a placeholder for missing values, stores boolean fraud indicators as Y/N strings, and keeps dates as unparsed text.

In this step, you will clean those issues, create derived columns that make analysis easier, and split the data into separate DataFrames that match the star schema you will build in the next step.

In this step, get ready to:
  • Clean the raw data by replacing placeholders, converting types, and parsing dates.
  • Standardize categorical values and create derived columns for severity tiers and age groups.
  • Split the cleaned DataFrame into dimension and fact DataFrames matching the star schema design.
Clean the raw data

The extraction step revealed that the dataset has three categories of data quality problems: placeholder values (? instead of proper nulls), string-encoded booleans (Y/N for fraud), and unparsed date strings. You will now write a transform() function that fixes all of these.

  • In etl_pipeline.py, find the end of the extract() function, then add the following code below it:
# --- Transform ---
def transform(df):
    """Clean, standardize, and reshape raw data for the star schema."""
    # Replace '?' placeholders with NaN
    df = df.replace("?", pd.NA)

    # Convert fraud_reported to boolean
    df["fraud_reported"] = df["fraud_reported"].map({"Y": True, "N": False})

    # Parse dates
    df["policy_bind_date"] = pd.to_datetime(df["policy_bind_date"])
    df["incident_date"] = pd.to_datetime(df["incident_date"])

What does this code do?

  • df.replace("?", pd.NA) swaps every ? string in the entire DataFrame with pandas' proper missing-value marker. This lets .isnull() detect them correctly.
  • .map({"Y": True, "N": False}) converts the string fraud indicator into a real Python boolean, which PostgreSQL can store as a BOOLEAN column.
  • pd.to_datetime() parses the date strings into proper datetime objects so PostgreSQL can store them as DATE columns and support date arithmetic.

You will not see a change yet. The next section adds the standardization logic to complete this function.

Standardize and derive new columns

Beyond fixing bad values, a good transform creates derived columns that make analysis easier. You will uppercase state codes for consistency, map text severity labels to numeric tiers, and bucket ages into groups.

  • Still inside the transform() function, add the following code directly below the incident_date line you just wrote:
    # Uppercase policy state
    df["policy_state"] = df["policy_state"].str.upper()

    # Create severity tier
    severity_map = {
        "Trivial Damage": 1,
        "Minor Damage": 2,
        "Major Damage": 3,
        "Total Loss": 4,
    }
    df["severity_tier"] = df["incident_severity"].map(severity_map)

    # Create age groups
    df["age_group"] = pd.cut(
        df["age"],
        bins=[17, 25, 40, 55, 100],
        labels=["Young", "Adult", "Middle", "Senior"],
    )

    print(f"\nTransformed: {df.shape[0]} rows")
    print(f"Fraud distribution:\n{df['fraud_reported'].value_counts()}")
    print(f"Severity tier distribution:\n{df['severity_tier'].value_counts().sort_index()}")

    return df

What does this code do?

  • .str.upper() standardizes state codes (e.g. oh becomes OH) so your queries can group by state without case mismatches.
  • severity_map converts text labels like Total Loss into numeric tiers (1-4). Numeric tiers let you sort and compare severity easily in SQL.
  • pd.cut() bins continuous ages into categorical groups. The bins [17, 25, 40, 55, 100] create four ranges: Young (18-25), Adult (26-40), Middle (41-55), Senior (56+).
  • Save etl_pipeline.py.
  • Confirm that your transform() function starts with def transform(df): and ends with return df. It should contain the cleaning, standardization, and print statements all within one function body.

Seeing indentation errors?

Every line inside transform() must be indented with exactly 4 spaces. If you mixed tabs and spaces, VS Code will underline the line in red. Make sure the second block of code you added is at the same indentation level as the first block (both inside the function).

Help me fix indentation errors in my transform function.

Split data into dimension and fact DataFrames

A star schema stores data in separate tables: dimension tables hold descriptive attributes (who, what, where) and a fact table holds the measurable events (claim amounts). You need to split your single cleaned DataFrame into five separate DataFrames that match this design.

What is dimensional modeling?

Dimensional modeling (also called the Kimball method) organizes data into fact tables (what happened) and dimension tables (context about what happened). Each row in the source maps to one row in each dimension and one row in the fact table.

Foreign keys in the fact table point back to each dimension. This is the pattern interviewers test for in data engineering roles.

  • In etl_pipeline.py, add the following function below your transform() function:
def build_dimensions(df):
    """Split transformed data into dimension and fact DataFrames."""
    dim_customer = df[
        [
            "age",
            "age_group",
            "insured_sex",
            "insured_education_level",
            "insured_occupation",
            "insured_hobbies",
            "insured_relationship",
            "capital-gains",
            "capital-loss",
        ]
    ].copy()
    dim_customer.columns = [
        "age",
        "age_group",
        "insured_sex",
        "insured_education_level",
        "insured_occupation",
        "insured_hobbies",
        "insured_relationship",
        "capital_gains",
        "capital_loss",
    ]
    dim_customer = dim_customer.reset_index(drop=True)
    dim_customer.index += 1
    dim_customer.index.name = "customer_id"

    dim_policy = df[
        [
            "policy_number",
            "policy_bind_date",
            "policy_state",
            "policy_csl",
            "policy_deductable",
            "policy_annual_premium",
            "umbrella_limit",
            "months_as_customer",
        ]
    ].copy()
    dim_policy = dim_policy.reset_index(drop=True)
    dim_policy.index += 1
    dim_policy.index.name = "policy_id"

    dim_vehicle = df[["auto_make", "auto_model", "auto_year"]].copy()
    dim_vehicle = dim_vehicle.reset_index(drop=True)
    dim_vehicle.index += 1
    dim_vehicle.index.name = "vehicle_id"

    dim_incident = df[
        [
            "incident_date",
            "incident_type",
            "collision_type",
            "incident_severity",
            "severity_tier",
            "authorities_contacted",
            "incident_state",
            "incident_city",
            "incident_hour_of_the_day",
            "number_of_vehicles_involved",
            "property_damage",
            "bodily_injuries",
            "witnesses",
            "police_report_available",
        ]
    ].copy()
    dim_incident = dim_incident.reset_index(drop=True)
    dim_incident.index += 1
    dim_incident.index.name = "incident_id"

    fact_claims = pd.DataFrame(
        {
            "customer_id": range(1, len(df) + 1),
            "policy_id": range(1, len(df) + 1),
            "vehicle_id": range(1, len(df) + 1),
            "incident_id": range(1, len(df) + 1),
            "total_claim_amount": df["total_claim_amount"].values,
            "injury_claim": df["injury_claim"].values,
            "property_claim": df["property_claim"].values,
            "vehicle_claim": df["vehicle_claim"].values,
            "fraud_reported": df["fraud_reported"].values,
        }
    )

    return dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims

What does this function do?

This function slices the cleaned DataFrame into five pieces that match the star schema tables you will create in the next step.

  • dim_customer holds insured demographics (age, sex, education, occupation).
  • dim_policy holds policy details (state, premium, deductible).
  • dim_vehicle holds auto info (make, model, year).
  • dim_incident holds incident details (type, severity, location).
  • fact_claims holds the measurable claim amounts plus foreign key IDs pointing to each dimension.

Each dimension DataFrame gets a 1-based index that will become its surrogate primary key in PostgreSQL. The capital-gains and capital-loss columns are renamed to use underscores because PostgreSQL does not allow hyphens in column names.

  • Save etl_pipeline.py.

Seeing a KeyError on a column name?

The raw CSV uses hyphens in capital-gains and capital-loss. Make sure you reference those columns with hyphens when selecting them from the DataFrame, then rename them to underscores afterwards. If you accidentally typed underscores in the selection list, pandas cannot find the column.

Help me debug a KeyError in my build_dimensions function.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py with both transform() and build_dimensions() below your existing extract() function.

ⓧ I'd like to double check the full code

"""Insurance Claims ETL Pipeline - Extract, Transform, Load into PostgreSQL Star Schema."""

import pandas as pd
from sqlalchemy import create_engine, text

# --- Configuration ---
DATABASE_URL = "postgresql+psycopg2://localhost/insurance_dw"
SOURCE_CSV = "data/insurance_claims.csv"


# --- Extract ---
def extract(filepath):
    """Extract raw data from CSV file."""
    df = pd.read_csv(filepath)
    print(f"Extracted: {df.shape[0]} rows, {df.shape[1]} columns")
    print(f"\nColumn types:\n{df.dtypes}")
    print(f"\nNull counts:\n{df.isnull().sum()[df.isnull().sum() > 0]}")
    print(f"\nPlaceholder '?' counts:")
    question_marks = (df == "?").sum()
    print(question_marks[question_marks > 0])
    return df


# --- Transform ---
def transform(df):
    """Clean, standardize, and reshape raw data for the star schema."""
    # Replace '?' placeholders with NaN
    df = df.replace("?", pd.NA)

    # Convert fraud_reported to boolean
    df["fraud_reported"] = df["fraud_reported"].map({"Y": True, "N": False})

    # Parse dates
    df["policy_bind_date"] = pd.to_datetime(df["policy_bind_date"])
    df["incident_date"] = pd.to_datetime(df["incident_date"])

    # Uppercase policy state
    df["policy_state"] = df["policy_state"].str.upper()

    # Create severity tier
    severity_map = {
        "Trivial Damage": 1,
        "Minor Damage": 2,
        "Major Damage": 3,
        "Total Loss": 4,
    }
    df["severity_tier"] = df["incident_severity"].map(severity_map)

    # Create age groups
    df["age_group"] = pd.cut(
        df["age"],
        bins=[17, 25, 40, 55, 100],
        labels=["Young", "Adult", "Middle", "Senior"],
    )

    print(f"\nTransformed: {df.shape[0]} rows")
    print(f"Fraud distribution:\n{df['fraud_reported'].value_counts()}")
    print(f"Severity tier distribution:\n{df['severity_tier'].value_counts().sort_index()}")

    return df


def build_dimensions(df):
    """Split transformed data into dimension and fact DataFrames."""
    dim_customer = df[
        [
            "age",
            "age_group",
            "insured_sex",
            "insured_education_level",
            "insured_occupation",
            "insured_hobbies",
            "insured_relationship",
            "capital-gains",
            "capital-loss",
        ]
    ].copy()
    dim_customer.columns = [
        "age",
        "age_group",
        "insured_sex",
        "insured_education_level",
        "insured_occupation",
        "insured_hobbies",
        "insured_relationship",
        "capital_gains",
        "capital_loss",
    ]
    dim_customer = dim_customer.reset_index(drop=True)
    dim_customer.index += 1
    dim_customer.index.name = "customer_id"

    dim_policy = df[
        [
            "policy_number",
            "policy_bind_date",
            "policy_state",
            "policy_csl",
            "policy_deductable",
            "policy_annual_premium",
            "umbrella_limit",
            "months_as_customer",
        ]
    ].copy()
    dim_policy = dim_policy.reset_index(drop=True)
    dim_policy.index += 1
    dim_policy.index.name = "policy_id"

    dim_vehicle = df[["auto_make", "auto_model", "auto_year"]].copy()
    dim_vehicle = dim_vehicle.reset_index(drop=True)
    dim_vehicle.index += 1
    dim_vehicle.index.name = "vehicle_id"

    dim_incident = df[
        [
            "incident_date",
            "incident_type",
            "collision_type",
            "incident_severity",
            "severity_tier",
            "authorities_contacted",
            "incident_state",
            "incident_city",
            "incident_hour_of_the_day",
            "number_of_vehicles_involved",
            "property_damage",
            "bodily_injuries",
            "witnesses",
            "police_report_available",
        ]
    ].copy()
    dim_incident = dim_incident.reset_index(drop=True)
    dim_incident.index += 1
    dim_incident.index.name = "incident_id"

    fact_claims = pd.DataFrame(
        {
            "customer_id": range(1, len(df) + 1),
            "policy_id": range(1, len(df) + 1),
            "vehicle_id": range(1, len(df) + 1),
            "incident_id": range(1, len(df) + 1),
            "total_claim_amount": df["total_claim_amount"].values,
            "injury_claim": df["injury_claim"].values,
            "property_claim": df["property_claim"].values,
            "vehicle_claim": df["vehicle_claim"].values,
            "fraud_reported": df["fraud_reported"].values,
        }
    )

    return dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims

Your transformation logic is ready. Next up, you will design the SQL schema that defines these tables in PostgreSQL and load the data in.

Design and Create the Star Schema

Your transformed DataFrames are ready, but they only live in Python's memory. To make this data queryable with SQL, you need to define the database structure and load everything into PostgreSQL.

In this step, you will write the DDL (Data Definition Language) that creates the star schema tables, execute it against your database, and load all 1,000 rows into each table using pandas.

In this step, get ready to:
  • Write the SQL schema file with dimension and fact table definitions.
  • Execute the schema against PostgreSQL using SQLAlchemy.
  • Load all dimension and fact DataFrames into the database.
Write the star schema DDL

A star schema organizes data into a central fact table (holding measurements like claim amounts) surrounded by dimension tables (holding descriptive attributes like customer demographics and vehicle details).

What is a star schema?

In the Kimball method of dimensional modeling, you separate measurable events (facts) from the context around them (dimensions). The fact table sits at the center and links to each dimension table via foreign keys.

This design makes analytical queries fast because you can filter and group by any dimension without scanning the entire dataset.

You will write the schema in three parts: first the dimension tables for customers and policies, then the remaining dimensions for vehicles and incidents, and finally the central fact table that ties everything together.

  • Open schema.sql in VS Code and add the DROP TABLE statements and first two dimension tables:
-- Star Schema for Insurance Claims Data Warehouse

DROP TABLE IF EXISTS fact_claims CASCADE;
DROP TABLE IF EXISTS dim_customer CASCADE;
DROP TABLE IF EXISTS dim_policy CASCADE;
DROP TABLE IF EXISTS dim_vehicle CASCADE;
DROP TABLE IF EXISTS dim_incident CASCADE;

CREATE TABLE dim_customer (
    customer_id SERIAL PRIMARY KEY,
    age INTEGER,
    age_group VARCHAR(20),
    insured_sex VARCHAR(10),
    insured_education_level VARCHAR(30),
    insured_occupation VARCHAR(50),
    insured_hobbies VARCHAR(50),
    insured_relationship VARCHAR(30),
    capital_gains INTEGER,
    capital_loss INTEGER
);

CREATE TABLE dim_policy (
    policy_id SERIAL PRIMARY KEY,
    policy_number INTEGER,
    policy_bind_date DATE,
    policy_state VARCHAR(5),
    policy_csl VARCHAR(20),
    policy_deductable INTEGER,
    policy_annual_premium NUMERIC(10,2),
    umbrella_limit INTEGER,
    months_as_customer INTEGER
);

What does this code do?

The DROP TABLE IF EXISTS ... CASCADE statements ensure you start fresh each time you run the schema. CASCADE removes any dependent objects (like foreign key references) so the drops succeed regardless of table order.

Each dimension table uses SERIAL PRIMARY KEY to auto-generate unique IDs. The column types match what your transformed DataFrames hold: integers, dates, varchars, and numerics.

  • Below the dim_policy table, add the vehicle and incident dimension tables:
CREATE TABLE dim_vehicle (
    vehicle_id SERIAL PRIMARY KEY,
    auto_make VARCHAR(30),
    auto_model VARCHAR(30),
    auto_year INTEGER
);

CREATE TABLE dim_incident (
    incident_id SERIAL PRIMARY KEY,
    incident_date DATE,
    incident_type VARCHAR(50),
    collision_type VARCHAR(50),
    incident_severity VARCHAR(30),
    severity_tier INTEGER,
    authorities_contacted VARCHAR(30),
    incident_state VARCHAR(5),
    incident_city VARCHAR(50),
    incident_hour_of_the_day INTEGER,
    number_of_vehicles_involved INTEGER,
    property_damage VARCHAR(5),
    bodily_injuries INTEGER,
    witnesses INTEGER,
    police_report_available VARCHAR(5)
);

What does this code do?

The dim_vehicle table is compact with just make, model, and year. The dim_incident table holds all incident context including the severity_tier column you derived during the transform step.

  • Finally, add the fact table that references all four dimension tables:
CREATE TABLE fact_claims (
    claim_id SERIAL PRIMARY KEY,
    customer_id INTEGER REFERENCES dim_customer(customer_id),
    policy_id INTEGER REFERENCES dim_policy(policy_id),
    vehicle_id INTEGER REFERENCES dim_vehicle(vehicle_id),
    incident_id INTEGER REFERENCES dim_incident(incident_id),
    total_claim_amount INTEGER,
    injury_claim INTEGER,
    property_claim INTEGER,
    vehicle_claim INTEGER,
    fraud_reported BOOLEAN
);

Why does fact_claims use REFERENCES?

The REFERENCES keyword creates foreign key constraints. PostgreSQL will reject any insert into fact_claims where a customer_id, policy_id, vehicle_id, or incident_id does not already exist in its corresponding dimension table.

This is why you must load dimension tables before the fact table. The constraints guarantee referential integrity at the database level.

  • Save schema.sql.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before continuing.

ⓧ I'd like to double check the full code

-- Star Schema for Insurance Claims Data Warehouse

DROP TABLE IF EXISTS fact_claims CASCADE;
DROP TABLE IF EXISTS dim_customer CASCADE;
DROP TABLE IF EXISTS dim_policy CASCADE;
DROP TABLE IF EXISTS dim_vehicle CASCADE;
DROP TABLE IF EXISTS dim_incident CASCADE;

CREATE TABLE dim_customer (
    customer_id SERIAL PRIMARY KEY,
    age INTEGER,
    age_group VARCHAR(20),
    insured_sex VARCHAR(10),
    insured_education_level VARCHAR(30),
    insured_occupation VARCHAR(50),
    insured_hobbies VARCHAR(50),
    insured_relationship VARCHAR(30),
    capital_gains INTEGER,
    capital_loss INTEGER
);

CREATE TABLE dim_policy (
    policy_id SERIAL PRIMARY KEY,
    policy_number INTEGER,
    policy_bind_date DATE,
    policy_state VARCHAR(5),
    policy_csl VARCHAR(20),
    policy_deductable INTEGER,
    policy_annual_premium NUMERIC(10,2),
    umbrella_limit INTEGER,
    months_as_customer INTEGER
);

CREATE TABLE dim_vehicle (
    vehicle_id SERIAL PRIMARY KEY,
    auto_make VARCHAR(30),
    auto_model VARCHAR(30),
    auto_year INTEGER
);

CREATE TABLE dim_incident (
    incident_id SERIAL PRIMARY KEY,
    incident_date DATE,
    incident_type VARCHAR(50),
    collision_type VARCHAR(50),
    incident_severity VARCHAR(30),
    severity_tier INTEGER,
    authorities_contacted VARCHAR(30),
    incident_state VARCHAR(5),
    incident_city VARCHAR(50),
    incident_hour_of_the_day INTEGER,
    number_of_vehicles_involved INTEGER,
    property_damage VARCHAR(5),
    bodily_injuries INTEGER,
    witnesses INTEGER,
    police_report_available VARCHAR(5)
);

CREATE TABLE fact_claims (
    claim_id SERIAL PRIMARY KEY,
    customer_id INTEGER REFERENCES dim_customer(customer_id),
    policy_id INTEGER REFERENCES dim_policy(policy_id),
    vehicle_id INTEGER REFERENCES dim_vehicle(vehicle_id),
    incident_id INTEGER REFERENCES dim_incident(incident_id),
    total_claim_amount INTEGER,
    injury_claim INTEGER,
    property_claim INTEGER,
    vehicle_claim INTEGER,
    fraud_reported BOOLEAN
);
Execute the schema against PostgreSQL

Now you need a Python function that reads schema.sql and executes it against the insurance_dw database. SQLAlchemy's engine.begin() context manager handles the connection and transaction automatically.

  • In etl_pipeline.py, find the end of the build_dimensions() function. Below it, add a new section comment and the create_schema() function:
# --- Load ---
def create_schema(engine):
    """Create the star schema tables in PostgreSQL."""
    with open("schema.sql", "r") as f:
        schema_sql = f.read()
    with engine.begin() as conn:
        conn.execute(text(schema_sql))
    print("Schema created successfully.")

What does this code do?

The function opens schema.sql and reads the entire file into a string. Then engine.begin() opens a database connection wrapped in a transaction.

The text() function from SQLAlchemy wraps the raw SQL string so the engine can execute it. If any statement fails, the transaction rolls back automatically.

  • Save etl_pipeline.py.
  • Test the schema creation by running this command in your terminal:
python -c "from etl_pipeline import create_schema; from sqlalchemy import create_engine; engine = create_engine('postgresql+psycopg2://localhost/insurance_dw'); create_schema(engine)"
  • You should see Schema created successfully. printed in the terminal.
  • Verify the tables exist by running this command:
psql insurance_dw -c "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_type = 'BASE TABLE';"
  • You should see all five tables listed: dim_customer, dim_policy, dim_vehicle, dim_incident, and fact_claims.

Schema creation failed?

If you see a connection error, make sure PostgreSQL is running. On macOS, check with brew services list or look for the PostgreSQL icon in your menu bar.

If you see a syntax error, open schema.sql and check for missing semicolons or mismatched parentheses.

Help me debug my schema creation error.

Load the data into PostgreSQL

With the tables created, you can now load your dimension DataFrames first (so their primary keys exist), then load the fact table that references them.

  • In etl_pipeline.py, add the load() function directly below create_schema():
def load(engine, dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims):
    """Load dimension and fact tables into PostgreSQL."""
    dim_customer.to_sql("dim_customer", engine, if_exists="append", index=True)
    print(f"Loaded dim_customer: {len(dim_customer)} rows")
    dim_policy.to_sql("dim_policy", engine, if_exists="append", index=True)
    print(f"Loaded dim_policy: {len(dim_policy)} rows")
    dim_vehicle.to_sql("dim_vehicle", engine, if_exists="append", index=True)
    print(f"Loaded dim_vehicle: {len(dim_vehicle)} rows")
    dim_incident.to_sql("dim_incident", engine, if_exists="append", index=True)
    print(f"Loaded dim_incident: {len(dim_incident)} rows")
    fact_claims.to_sql("fact_claims", engine, if_exists="append", index=False)
    print(f"Loaded fact_claims: {len(fact_claims)} rows")

What does this code do?

The to_sql() method from pandas writes a DataFrame directly into a PostgreSQL table. The if_exists="append" argument adds rows to the existing table (which create_schema() already created) rather than replacing it.

Dimension tables use index=True because their DataFrame index IS the primary key (e.g. customer_id). The fact table uses index=False because its claim_id is auto-generated by the SERIAL column in PostgreSQL.

  • Save etl_pipeline.py.
  • Run the full Extract-Transform-Load pipeline by running this command:
python -c "
from etl_pipeline import *
engine = create_engine(DATABASE_URL)
raw_df = extract(SOURCE_CSV)
cleaned_df = transform(raw_df)
dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims = build_dimensions(cleaned_df)
create_schema(engine)
load(engine, dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims)
"
  • You should see output confirming 1000 rows extracted, transformation stats, schema creation, and each table loaded with 1000 rows.
  • Verify the row counts in PostgreSQL by running:
psql insurance_dw -c "SELECT 'dim_customer' AS tbl, COUNT(*) FROM dim_customer UNION ALL SELECT 'dim_policy', COUNT(*) FROM dim_policy UNION ALL SELECT 'dim_vehicle', COUNT(*) FROM dim_vehicle UNION ALL SELECT 'dim_incident', COUNT(*) FROM dim_incident UNION ALL SELECT 'fact_claims', COUNT(*) FROM fact_claims;"
  • You should see 1000 rows in every table.

Seeing a foreign key violation?

If you see an error about a foreign key constraint, the dimension tables did not load before the fact table. Make sure create_schema(engine) ran first to reset the tables, then load() processes dimensions before fact_claims.

If you see a duplicate key error, the tables already had data from a previous run. Run the pipeline again from scratch. The DROP TABLE statements in schema.sql will clear everything.

Help me fix my data loading error.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py before continuing.

ⓧ I'd like to double check the full code

"""Insurance Claims ETL Pipeline - Extract, Transform, Load into PostgreSQL Star Schema."""

import pandas as pd
from sqlalchemy import create_engine, text

# --- Configuration ---
DATABASE_URL = "postgresql+psycopg2://localhost/insurance_dw"
SOURCE_CSV = "data/insurance_claims.csv"


# --- Extract ---
def extract(filepath):
    """Extract raw data from CSV file."""
    df = pd.read_csv(filepath)
    print(f"Extracted: {df.shape[0]} rows, {df.shape[1]} columns")
    print(f"\nColumn types:\n{df.dtypes}")
    print(f"\nNull counts:\n{df.isnull().sum()[df.isnull().sum() > 0]}")
    print(f"\nPlaceholder '?' counts:")
    question_marks = (df == "?").sum()
    print(question_marks[question_marks > 0])
    return df


# --- Transform ---
def transform(df):
    """Clean, standardize, and reshape raw data for the star schema."""
    df = df.replace("?", pd.NA)
    df["fraud_reported"] = df["fraud_reported"].map({"Y": True, "N": False})
    df["policy_bind_date"] = pd.to_datetime(df["policy_bind_date"])
    df["incident_date"] = pd.to_datetime(df["incident_date"])
    df["policy_state"] = df["policy_state"].str.upper()
    severity_map = {
        "Trivial Damage": 1,
        "Minor Damage": 2,
        "Major Damage": 3,
        "Total Loss": 4,
    }
    df["severity_tier"] = df["incident_severity"].map(severity_map)
    df["age_group"] = pd.cut(
        df["age"],
        bins=[17, 25, 40, 55, 100],
        labels=["Young", "Adult", "Middle", "Senior"],
    )
    print(f"\nTransformed: {df.shape[0]} rows")
    print(f"Fraud distribution:\n{df['fraud_reported'].value_counts()}")
    print(f"Severity tier distribution:\n{df['severity_tier'].value_counts().sort_index()}")
    return df


def build_dimensions(df):
    """Split transformed data into dimension and fact DataFrames."""
    dim_customer = df[["age","age_group","insured_sex","insured_education_level","insured_occupation","insured_hobbies","insured_relationship","capital-gains","capital-loss"]].copy()
    dim_customer.columns = ["age","age_group","insured_sex","insured_education_level","insured_occupation","insured_hobbies","insured_relationship","capital_gains","capital_loss"]
    dim_customer = dim_customer.reset_index(drop=True)
    dim_customer.index += 1
    dim_customer.index.name = "customer_id"

    dim_policy = df[["policy_number","policy_bind_date","policy_state","policy_csl","policy_deductable","policy_annual_premium","umbrella_limit","months_as_customer"]].copy()
    dim_policy = dim_policy.reset_index(drop=True)
    dim_policy.index += 1
    dim_policy.index.name = "policy_id"

    dim_vehicle = df[["auto_make","auto_model","auto_year"]].copy()
    dim_vehicle = dim_vehicle.reset_index(drop=True)
    dim_vehicle.index += 1
    dim_vehicle.index.name = "vehicle_id"

    dim_incident = df[["incident_date","incident_type","collision_type","incident_severity","severity_tier","authorities_contacted","incident_state","incident_city","incident_hour_of_the_day","number_of_vehicles_involved","property_damage","bodily_injuries","witnesses","police_report_available"]].copy()
    dim_incident = dim_incident.reset_index(drop=True)
    dim_incident.index += 1
    dim_incident.index.name = "incident_id"

    fact_claims = pd.DataFrame({
        "customer_id": range(1, len(df) + 1),
        "policy_id": range(1, len(df) + 1),
        "vehicle_id": range(1, len(df) + 1),
        "incident_id": range(1, len(df) + 1),
        "total_claim_amount": df["total_claim_amount"].values,
        "injury_claim": df["injury_claim"].values,
        "property_claim": df["property_claim"].values,
        "vehicle_claim": df["vehicle_claim"].values,
        "fraud_reported": df["fraud_reported"].values,
    })

    return dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims


# --- Load ---
def create_schema(engine):
    """Create the star schema tables in PostgreSQL."""
    with open("schema.sql", "r") as f:
        schema_sql = f.read()
    with engine.begin() as conn:
        conn.execute(text(schema_sql))
    print("Schema created successfully.")


def load(engine, dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims):
    """Load dimension and fact tables into PostgreSQL."""
    dim_customer.to_sql("dim_customer", engine, if_exists="append", index=True)
    print(f"Loaded dim_customer: {len(dim_customer)} rows")
    dim_policy.to_sql("dim_policy", engine, if_exists="append", index=True)
    print(f"Loaded dim_policy: {len(dim_policy)} rows")
    dim_vehicle.to_sql("dim_vehicle", engine, if_exists="append", index=True)
    print(f"Loaded dim_vehicle: {len(dim_vehicle)} rows")
    dim_incident.to_sql("dim_incident", engine, if_exists="append", index=True)
    print(f"Loaded dim_incident: {len(dim_incident)} rows")
    fact_claims.to_sql("fact_claims", engine, if_exists="append", index=False)
    print(f"Loaded fact_claims: {len(fact_claims)} rows")

Your data warehouse is live. Next up, you will write analytical queries that prove this star schema can answer real insurance business questions.

Run Analytical Queries

Your warehouse is loaded with clean, structured data across four dimension tables and one fact table. The whole point of building a data warehouse is to answer business questions quickly.

In this step, you will write SQL queries that join your fact and dimension tables to surface insurance insights. You will also wire up the main block that runs the entire ETL pipeline end-to-end.

In this step, get ready to:
  • Write analytical queries for fraud rates, claim severity, and top auto makes.
  • Run a multi-join risk segmentation query across customers, policies, and claims.
  • Add the main block and run the full pipeline end-to-end.
Write business queries for fraud, severity, and top makes

The star schema lets you answer business questions by joining the central fact table to whichever dimension holds the grouping column. Each query below joins fact_claims to a different dimension.

  • In etl_pipeline.py, add the following run_queries() function below the load() function:
# --- Analyze ---
def run_queries(engine):
    """Run analytical queries against the warehouse."""
    print("\n--- Analytical Queries ---\n")

    # Query 1: Fraud rate by policy state
    query_fraud = """
        SELECT dp.policy_state,
               COUNT(*) AS total_claims,
               SUM(CASE WHEN fc.fraud_reported THEN 1 ELSE 0 END) AS fraud_count,
               ROUND(100.0 * SUM(CASE WHEN fc.fraud_reported THEN 1 ELSE 0 END) / COUNT(*), 1) AS fraud_rate_pct
        FROM fact_claims fc
        JOIN dim_policy dp ON fc.policy_id = dp.policy_id
        GROUP BY dp.policy_state
        ORDER BY fraud_rate_pct DESC;
    """
    print("1. Fraud Rate by Policy State:")
    print(pd.read_sql(query_fraud, engine))

    # Query 2: Average claim amount by severity tier
    query_severity = """
        SELECT di.incident_severity,
               di.severity_tier,
               COUNT(*) AS claim_count,
               ROUND(AVG(fc.total_claim_amount), 2) AS avg_claim
        FROM fact_claims fc
        JOIN dim_incident di ON fc.incident_id = di.incident_id
        GROUP BY di.incident_severity, di.severity_tier
        ORDER BY di.severity_tier;
    """
    print("\n2. Average Claim Amount by Severity:")
    print(pd.read_sql(query_severity, engine))

    # Query 3: Top 5 auto makes by total claim value
    query_makes = """
        SELECT dv.auto_make,
               COUNT(*) AS claim_count,
               SUM(fc.total_claim_amount) AS total_claims_value
        FROM fact_claims fc
        JOIN dim_vehicle dv ON fc.vehicle_id = dv.vehicle_id
        GROUP BY dv.auto_make
        ORDER BY total_claims_value DESC
        LIMIT 5;
    """
    print("\n3. Top 5 Auto Makes by Total Claim Value:")
    print(pd.read_sql(query_makes, engine))

What do these queries do?

  • query_fraud joins fact_claims to dim_policy and calculates fraud percentage per state using a conditional SUM.
  • query_severity joins to dim_incident and groups by the severity tier you created during transformation to show average claim amounts.
  • query_makes joins to dim_vehicle and ranks car manufacturers by total claim dollars.

Each query uses pd.read_sql() which sends the SQL to PostgreSQL and returns the results as a pandas DataFrame.

  • Save etl_pipeline.py.

You won't see a change yet. The next chunk adds a risk segmentation query, and then you will run everything together.

Add the risk segmentation query

The three queries above each join one dimension. A risk segmentation query joins multiple dimensions at once to find high-risk customer segments.

  • In etl_pipeline.py, add the following query at the bottom of the run_queries() function, below the print(pd.read_sql(query_makes, engine)) line:
    # Query 4: Risk segmentation across customers and policies
    query_risk = """
        SELECT dc.age_group,
               dp.policy_state,
               COUNT(*) AS claim_count,
               ROUND(AVG(fc.total_claim_amount), 2) AS avg_claim
        FROM fact_claims fc
        JOIN dim_customer dc ON fc.customer_id = dc.customer_id
        JOIN dim_policy dp ON fc.policy_id = dp.policy_id
        GROUP BY dc.age_group, dp.policy_state
        HAVING COUNT(*) >= 10
        ORDER BY avg_claim DESC
        LIMIT 10;
    """
    print("\n4. High-Risk Segments (Age Group + State):")
    print(pd.read_sql(query_risk, engine))

Why is this query more powerful?

This query joins two dimension tables (dim_customer and dim_policy) to fact_claims at the same time. This multi-dimensional grouping reveals which age group and state combinations file the highest average claims.

The HAVING COUNT(*) >= 10 filter removes segments with too few claims to be statistically meaningful.

  • Save etl_pipeline.py.

You won't see results yet. The next chunk adds the main block that runs the full pipeline.

Add the main block and run the full pipeline

The main block ties every stage together: extract, transform, build dimensions, create schema, load, and query. Running it executes the complete ETL pipeline in one command.

  • At the very bottom of etl_pipeline.py, below the run_queries() function, add the main block:
# --- Main ---
if __name__ == "__main__":
    # Extract
    raw_df = extract(SOURCE_CSV)

    # Transform
    cleaned_df = transform(raw_df)
    dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims = build_dimensions(
        cleaned_df
    )

    # Load
    engine = create_engine(DATABASE_URL)
    create_schema(engine)
    load(engine, dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims)

    # Analyze
    run_queries(engine)

    print("\nETL pipeline complete.")

What does the main block do?

  • if __name__ == "__main__": ensures this code only runs when you execute the script directly, not when it is imported as a module.
  • Each ETL stage is called in order: extract the CSV, transform and split it, create the schema, load the tables, then run analytical queries.
  • create_schema(engine) drops and recreates tables on every run, so you can safely re-run the pipeline without duplicate data.
  • Save etl_pipeline.py.
  • Run the full pipeline by executing this command in your terminal:
python etl_pipeline.py

You should see output for each ETL stage, followed by four formatted DataFrames showing fraud rates by state, average claims by severity, top auto makes, and high-risk segments. The load stage should confirm 1000 rows in fact_claims.

Seeing a connection error?

  • Make sure PostgreSQL is running. If you stopped it since the last step, start it again before running the script.
  • Check that your DATABASE_URL in etl_pipeline.py matches your local PostgreSQL configuration. If your PostgreSQL requires a username and password, update the connection string to include them.
  • If you see a "relation does not exist" error, confirm the schema.sql file is in the same directory as etl_pipeline.py.

Help me debug my ETL pipeline connection error.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl_pipeline.py and that your terminal output shows all four query results plus the final "ETL pipeline complete." message.

ⓧ I'd like to double check the full code

"""Insurance Claims ETL Pipeline - Extract, Transform, Load into PostgreSQL Star Schema."""

import pandas as pd
from sqlalchemy import create_engine, text

# --- Configuration ---
DATABASE_URL = "postgresql+psycopg2://localhost/insurance_dw"
SOURCE_CSV = "data/insurance_claims.csv"


# --- Extract ---
def extract(filepath):
    """Extract raw data from CSV file."""
    df = pd.read_csv(filepath)
    print(f"Extracted: {df.shape[0]} rows, {df.shape[1]} columns")
    print(f"\nColumn types:\n{df.dtypes}")
    print(f"\nNull counts:\n{df.isnull().sum()[df.isnull().sum() > 0]}")
    print(f"\nPlaceholder '?' counts:")
    question_marks = (df == "?").sum()
    print(question_marks[question_marks > 0])
    return df


# --- Transform ---
def transform(df):
    """Clean, standardize, and reshape raw data for the star schema."""
    df = df.replace("?", pd.NA)
    df["fraud_reported"] = df["fraud_reported"].map({"Y": True, "N": False})
    df["policy_bind_date"] = pd.to_datetime(df["policy_bind_date"])
    df["incident_date"] = pd.to_datetime(df["incident_date"])
    df["policy_state"] = df["policy_state"].str.upper()
    severity_map = {
        "Trivial Damage": 1,
        "Minor Damage": 2,
        "Major Damage": 3,
        "Total Loss": 4,
    }
    df["severity_tier"] = df["incident_severity"].map(severity_map)
    df["age_group"] = pd.cut(
        df["age"],
        bins=[17, 25, 40, 55, 100],
        labels=["Young", "Adult", "Middle", "Senior"],
    )
    print(f"\nTransformed: {df.shape[0]} rows")
    print(f"Fraud distribution:\n{df['fraud_reported'].value_counts()}")
    print(f"Severity tier distribution:\n{df['severity_tier'].value_counts().sort_index()}")
    return df


def build_dimensions(df):
    """Split transformed data into dimension and fact DataFrames."""
    dim_customer = df[["age","age_group","insured_sex","insured_education_level","insured_occupation","insured_hobbies","insured_relationship","capital-gains","capital-loss"]].copy()
    dim_customer.columns = ["age","age_group","insured_sex","insured_education_level","insured_occupation","insured_hobbies","insured_relationship","capital_gains","capital_loss"]
    dim_customer = dim_customer.reset_index(drop=True)
    dim_customer.index += 1
    dim_customer.index.name = "customer_id"

    dim_policy = df[["policy_number","policy_bind_date","policy_state","policy_csl","policy_deductable","policy_annual_premium","umbrella_limit","months_as_customer"]].copy()
    dim_policy = dim_policy.reset_index(drop=True)
    dim_policy.index += 1
    dim_policy.index.name = "policy_id"

    dim_vehicle = df[["auto_make","auto_model","auto_year"]].copy()
    dim_vehicle = dim_vehicle.reset_index(drop=True)
    dim_vehicle.index += 1
    dim_vehicle.index.name = "vehicle_id"

    dim_incident = df[["incident_date","incident_type","collision_type","incident_severity","severity_tier","authorities_contacted","incident_state","incident_city","incident_hour_of_the_day","number_of_vehicles_involved","property_damage","bodily_injuries","witnesses","police_report_available"]].copy()
    dim_incident = dim_incident.reset_index(drop=True)
    dim_incident.index += 1
    dim_incident.index.name = "incident_id"

    fact_claims = pd.DataFrame({
        "customer_id": range(1, len(df) + 1),
        "policy_id": range(1, len(df) + 1),
        "vehicle_id": range(1, len(df) + 1),
        "incident_id": range(1, len(df) + 1),
        "total_claim_amount": df["total_claim_amount"].values,
        "injury_claim": df["injury_claim"].values,
        "property_claim": df["property_claim"].values,
        "vehicle_claim": df["vehicle_claim"].values,
        "fraud_reported": df["fraud_reported"].values,
    })

    return dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims


# --- Load ---
def create_schema(engine):
    """Create the star schema tables in PostgreSQL."""
    with open("schema.sql", "r") as f:
        schema_sql = f.read()
    with engine.begin() as conn:
        conn.execute(text(schema_sql))
    print("Schema created successfully.")


def load(engine, dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims):
    """Load dimension and fact tables into PostgreSQL."""
    dim_customer.to_sql("dim_customer", engine, if_exists="append", index=True)
    print(f"Loaded dim_customer: {len(dim_customer)} rows")
    dim_policy.to_sql("dim_policy", engine, if_exists="append", index=True)
    print(f"Loaded dim_policy: {len(dim_policy)} rows")
    dim_vehicle.to_sql("dim_vehicle", engine, if_exists="append", index=True)
    print(f"Loaded dim_vehicle: {len(dim_vehicle)} rows")
    dim_incident.to_sql("dim_incident", engine, if_exists="append", index=True)
    print(f"Loaded dim_incident: {len(dim_incident)} rows")
    fact_claims.to_sql("fact_claims", engine, if_exists="append", index=False)
    print(f"Loaded fact_claims: {len(fact_claims)} rows")


# --- Analyze ---
def run_queries(engine):
    """Run analytical queries against the warehouse."""
    print("\n--- Analytical Queries ---\n")

    query_fraud = """
        SELECT dp.policy_state,
               COUNT(*) AS total_claims,
               SUM(CASE WHEN fc.fraud_reported THEN 1 ELSE 0 END) AS fraud_count,
               ROUND(100.0 * SUM(CASE WHEN fc.fraud_reported THEN 1 ELSE 0 END) / COUNT(*), 1) AS fraud_rate_pct
        FROM fact_claims fc
        JOIN dim_policy dp ON fc.policy_id = dp.policy_id
        GROUP BY dp.policy_state
        ORDER BY fraud_rate_pct DESC;
    """
    print("1. Fraud Rate by Policy State:")
    print(pd.read_sql(query_fraud, engine))

    query_severity = """
        SELECT di.incident_severity,
               di.severity_tier,
               COUNT(*) AS claim_count,
               ROUND(AVG(fc.total_claim_amount), 2) AS avg_claim
        FROM fact_claims fc
        JOIN dim_incident di ON fc.incident_id = di.incident_id
        GROUP BY di.incident_severity, di.severity_tier
        ORDER BY di.severity_tier;
    """
    print("\n2. Average Claim Amount by Severity:")
    print(pd.read_sql(query_severity, engine))

    query_makes = """
        SELECT dv.auto_make,
               COUNT(*) AS claim_count,
               SUM(fc.total_claim_amount) AS total_claims_value
        FROM fact_claims fc
        JOIN dim_vehicle dv ON fc.vehicle_id = dv.vehicle_id
        GROUP BY dv.auto_make
        ORDER BY total_claims_value DESC
        LIMIT 5;
    """
    print("\n3. Top 5 Auto Makes by Total Claim Value:")
    print(pd.read_sql(query_makes, engine))

    query_risk = """
        SELECT dc.age_group,
               dp.policy_state,
               COUNT(*) AS claim_count,
               ROUND(AVG(fc.total_claim_amount), 2) AS avg_claim
        FROM fact_claims fc
        JOIN dim_customer dc ON fc.customer_id = dc.customer_id
        JOIN dim_policy dp ON fc.policy_id = dp.policy_id
        GROUP BY dc.age_group, dp.policy_state
        HAVING COUNT(*) >= 10
        ORDER BY avg_claim DESC
        LIMIT 10;
    """
    print("\n4. High-Risk Segments (Age Group + State):")
    print(pd.read_sql(query_risk, engine))


# --- Main ---
if __name__ == "__main__":
    raw_df = extract(SOURCE_CSV)
    cleaned_df = transform(raw_df)
    dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims = build_dimensions(
        cleaned_df
    )
    engine = create_engine(DATABASE_URL)
    create_schema(engine)
    load(engine, dim_customer, dim_policy, dim_vehicle, dim_incident, fact_claims)
    run_queries(engine)
    print("\nETL pipeline complete.")

Your ETL pipeline is complete. You have gone from a raw CSV file all the way to formatted analytical results that answer real insurance business questions. Next up, you will take this pipeline from prototype to production by adding data quality checks in the Secret Mission.

Secret mission

Add Data Quality Validation

Your pipeline works, but how do you know the data arrived correctly? Production pipelines include automated quality checks that catch silent failures before bad data reaches analysts. In this mission, you will add row count reconciliation, null audits, and referential integrity validation to make your ETL pipeline production-grade.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • PostgreSQL database insurance_dw (local, no cost).
  • Python virtual environment with installed packages (pandas, sqlalchemy, psycopg2-binary).
  • Project files (etl_pipeline.py, schema.sql, data/insurance_claims.csv).

Keep everything running

No action needed. Choose this if you're still actively building or want to keep the warehouse for portfolio demos.

  • The insurance_dw database stays on your local PostgreSQL, consuming minimal disk space.
  • Your virtual environment and project files remain in your project folder on the Desktop.
  • You can re-run the pipeline at any time with python etl_pipeline.py.

Pause - I'll come back to this later

Since everything is local, there is nothing to pause. Your database and files will wait for you exactly as they are.

  • Deactivate your virtual environment by running:
deactivate
  • When you come back, reactivate it with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

Delete - I don't want to use this again

Remove all project resources and start fresh. This drops the database and deletes all local files.

  • Drop the PostgreSQL database by running this in your terminal:
psql postgres -c "DROP DATABASE insurance_dw;"
  • Delete the project folder (including the virtual environment and all files) by running:

macOS

cd ~/Desktop
rm -rf insurance-dw

Windows

cd ~\Desktop
Remove-Item -Recurse -Force insurance-dw
  • Confirm the database is gone by running:
psql postgres -c "\l"

You should no longer see insurance_dw in the list of databases.

Nice Work!

Nice Work!

Nice work! You just built a complete data engineering pipeline from scratch, taking raw insurance claims data all the way from a messy CSV into a properly modeled analytical data warehouse.

You've learned how to:

  • Build a Python ETL pipeline that extracts raw CSV data, profiles and cleans it (replacing placeholder values, parsing dates, converting types), and loads it into a relational database.
  • Design a Kimball-style star schema with four dimension tables and one fact table, connected by foreign key relationships in PostgreSQL.
  • Run analytical SQL queries against the warehouse to answer real business questions like fraud rates, severity analysis, and risk segmentation.
  • Secret Mission: Added production-grade data quality validation checks including row count reconciliation, null audits, and referential integrity verification.

Ready to quiz yourself?