Master Data Management Onboarding
Build a fuzzy-matching MDM pipeline to onboard acquisition data into a warehouse.
Introduction
30 Second Summary
When a company acquires another business, the hardest part is not the deal itself. It is merging two separate sets of customer and product data without creating duplicates that break every report downstream.
In this project, you will build a Master Data Management pipeline that onboards a new acquisition's messy data into your existing BalticDW warehouse using entity resolution, survivorship rules, and data quality gates.
What You'll Build
You will feed UAB Vilandra's customer and product records into your pipeline and watch it automatically resolve "Baltic Fuel SpA" to your existing "Baltic Fuel" record, quarantine borderline matches for human review, and reject garbage. Then you will run dbt build and see Vilandra's sales flow through your reporting marts as if they had always been part of the group.
By the end of this project, you'll have:
- A fuzzy matching engine where you feed it messy acquisition records and watch it classify each as auto-match, quarantine, or reject based on configurable similarity scores.
- Per-attribute survivorship rules that automatically decide which source wins for each field, with a full audit trail you can query to explain every merge decision.
- A data quality gate that quarantines ambiguous records and rejects invalid data before it touches your production warehouse, then end-to-end proof that Vilandra's sales flow cleanly through dbt into your reporting layer.
- Secret Mission: Build an interactive data steward review workflow that presents quarantined records side-by-side with their candidate matches and lets you approve or reject each one from the terminal.
Are there any prerequisites?
This is Project 4 in the JNG Baltic data platform series. You need a completed BalticDW from Projects 1-3 (star schema, dbt transformation layer, and Power BI dashboard).
You also need Python 3.14+, SQL Server 2025, ODBC Driver 18, dbt-core 1.12.0, and VS Code installed from the earlier projects.
Before We Start
Before We Start...
Before we dive in, let's lock in what you are about to build and why it matters for the JNG data platform.
Set Up the MDM Environment
Your BalticDW has clean, canonical dimensions from Projects 1-3. Customers, products, subsidiaries, and dates all sit in well-defined star schema tables with dbt tests guarding data quality.
But the warehouse has no mechanism to ingest a new acquisition's data safely. If UAB Vilandra's export lands directly in your production tables, you will get duplicates, broken keys, and failing reports. Before any new data touches the production schema, you need a dedicated staging area and a fuzzy matching library.
In this step, get ready to:
- Install RapidFuzz, the fuzzy string matching library that powers entity resolution.
- Create the mdm schema with staging, quarantine, and audit tables in BalticDW.
- Generate simulated UAB Vilandra customer and product data as CSV files.
Install RapidFuzz
Entity resolution compares incoming strings against existing records to find matches. RapidFuzz is a high-performance fuzzy string matching library written in C++ with Python bindings. It scores how similar two strings are on a 0-100 scale.
- Install RapidFuzz by running this command in your terminal:
pip install rapidfuzzWhat does this install?
RapidFuzz provides scorers like fuzz.ratio and fuzz.token_sort_ratio that calculate how similar two strings are. It also provides process.extractOne which finds the single best match for a query string within a list of candidates.
- Verify the installation by running:
python -c "from rapidfuzz import fuzz; print(fuzz.ratio('this is a test', 'this is a test!'))"You should see 96.55172413793103 printed. This confirms RapidFuzz is installed and its similarity scoring works.
Seeing an ImportError about DLL load failed?
On Windows, RapidFuzz requires the Visual C++ 2019 redistributable. Your SQL Server 2025 installation likely installed this already, but if you see a DLL error, download and install the redistributable from the Microsoft support page.
Help me fix this RapidFuzz import error
Create the mdm schema and tables in BalticDW
The MDM layer sits between raw acquisition data and your production dbo schema. It provides a safe landing zone where incoming records are staged, matched, quarantined if ambiguous, and audited before they touch production.
- Create a new file called setup_mdm_schema.sql in your project directory.
- Add the schema creation and staging customers table:
-- Create the MDM schema and tables in BalticDW
USE BalticDW;
GO
-- Create mdm schema if it does not exist
IF NOT EXISTS (SELECT 1 FROM sys.schemas WHERE name = 'mdm')
EXEC('CREATE SCHEMA mdm');
GO
-- Staging table for incoming customer records
IF OBJECT_ID('mdm.staging_customers', 'U') IS NOT NULL
DROP TABLE mdm.staging_customers;
CREATE TABLE mdm.staging_customers (
staging_id INT IDENTITY(1,1) PRIMARY KEY,
source_system NVARCHAR(50) NOT NULL,
customer_id NVARCHAR(50),
customer_name NVARCHAR(200),
segment NVARCHAR(50),
city NVARCHAR(100),
country_code NVARCHAR(10),
loaded_at DATETIME2 DEFAULT SYSUTCDATETIME(),
status NVARCHAR(20) DEFAULT 'PENDING'
);
GOWhat does this code do?
The mdm schema creates a namespace that isolates MDM infrastructure from your production dbo tables and your dbt_dev transformation layer.
The staging_customers table is the landing zone for incoming acquisition data. Each row tracks its source system, load time, and processing status so nothing gets lost or double-processed.
- Add the staging products table below the previous code:
-- Staging table for incoming product records
IF OBJECT_ID('mdm.staging_products', 'U') IS NOT NULL
DROP TABLE mdm.staging_products;
CREATE TABLE mdm.staging_products (
staging_id INT IDENTITY(1,1) PRIMARY KEY,
source_system NVARCHAR(50) NOT NULL,
product_id NVARCHAR(50),
product_name NVARCHAR(200),
category NVARCHAR(100),
brand NVARCHAR(100),
loaded_at DATETIME2 DEFAULT SYSUTCDATETIME(),
status NVARCHAR(20) DEFAULT 'PENDING'
);
GO- Open a SQL query window connected to BalticDW in SSMS or VS Code.
- Paste and execute the code you have so far.
You should see "Commands completed successfully" in the messages pane. This confirms the schema and both staging tables are created.
- Now add the quarantine table to your file:
-- Quarantine table for ambiguous matches
IF OBJECT_ID('mdm.quarantine', 'U') IS NOT NULL
DROP TABLE mdm.quarantine;
CREATE TABLE mdm.quarantine (
quarantine_id INT IDENTITY(1,1) PRIMARY KEY,
source_system NVARCHAR(50) NOT NULL,
entity_type NVARCHAR(20) NOT NULL,
source_id NVARCHAR(50),
source_name NVARCHAR(200),
candidate_key INT,
candidate_name NVARCHAR(200),
match_score FLOAT,
status NVARCHAR(20) DEFAULT 'PENDING',
resolved_by NVARCHAR(100),
resolved_at DATETIME2,
quarantined_at DATETIME2 DEFAULT SYSUTCDATETIME()
);
GOWhy a quarantine table?
When a fuzzy match scores between 70 and 89, the system is not confident enough to auto-merge but the record is too similar to ignore. The quarantine table holds these ambiguous records for human review, tracking which candidate they almost matched and what score they received.
- Add the merge audit log and the final PRINT statement:
-- Merge audit log
IF OBJECT_ID('mdm.merge_log', 'U') IS NOT NULL
DROP TABLE mdm.merge_log;
CREATE TABLE mdm.merge_log (
log_id INT IDENTITY(1,1) PRIMARY KEY,
entity_type NVARCHAR(20) NOT NULL,
entity_key INT NOT NULL,
source_system NVARCHAR(50) NOT NULL,
source_id NVARCHAR(50),
attribute_name NVARCHAR(100) NOT NULL,
old_value NVARCHAR(500),
new_value NVARCHAR(500),
survivorship_rule NVARCHAR(50) NOT NULL,
steward_override BIT DEFAULT 0,
merged_at DATETIME2 DEFAULT SYSUTCDATETIME()
);
GO
PRINT 'MDM schema and tables created successfully.';What is the merge log for?
Every attribute-level merge decision is recorded here: what the old value was, what the new value is, and which survivorship rule made the call. This gives you full audit trail provenance per golden attribute.
- Execute the full file again (including the quarantine and merge_log sections you just added).
You should see "MDM schema and tables created successfully." printed in the messages pane.
- Confirm all four tables exist by running this query:
SELECT TABLE_SCHEMA, TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = 'mdm'
ORDER BY TABLE_NAME;You should see four rows: merge_log, quarantine, staging_customers, and staging_products.
Not seeing the mdm tables?
Make sure your query window is connected to the BalticDW database, not master. Check the database dropdown in SSMS or the connection info in VS Code.
If you see an error about permissions on CREATE SCHEMA, confirm you are connected with an account that has db_owner rights on BalticDW (which you set up in Project 1).
Help me troubleshoot my mdm schema setup
Generate the simulated UAB Vilandra data
A real acquisition export would arrive as a data dump with inconsistencies, overlaps, and garbage. You will simulate this with a Python script that generates deterministic test data designed to produce specific outcomes in the matching pipeline.
- Create a new file called generate_vilandra.py in your project directory.
- Add the imports and the customer generation function:
"""Generate simulated UAB Vilandra acquisition data for MDM testing."""
import csv
import os
def generate_customers():
"""Create Vilandra customer CSV with realistic overlaps and new entities."""
customers = [
# Overlaps with existing BalticDW customers (fuzzy matches)
{"customer_id": "VIL-C001", "customer_name": "Baltic Fuel SpA", "segment": "Retail", "city": "Vilnius", "country_code": "LT"},
{"customer_id": "VIL-C002", "customer_name": "Maxima Group", "segment": "Retail", "city": "Kaunas", "country_code": "LT"},
{"customer_id": "VIL-C003", "customer_name": "Rimi Baltic Trading", "segment": "HORECA", "city": "Vilnius", "country_code": "LT"},
# Borderline matches (quarantine candidates, score 70-89)
{"customer_id": "VIL-C004", "customer_name": "Tallink Hotels Group", "segment": "HORECA", "city": "Vilnius", "country_code": "LT"},
{"customer_id": "VIL-C005", "customer_name": "Circle-K Baltics", "segment": "Retail", "city": "Klaipeda", "country_code": "LT"},
# Genuinely new customers (no match expected)
{"customer_id": "VIL-C006", "customer_name": "Vilandra Beverages OOO", "segment": "Standard", "city": "Vilnius", "country_code": "LT"},
{"customer_id": "VIL-C007", "customer_name": "Panevezys Distribution", "segment": "Standard", "city": "Panevezys", "country_code": "LT"},
{"customer_id": "VIL-C008", "customer_name": "Kauno Tiekimas UAB", "segment": "Premium", "city": "Kaunas", "country_code": "LT"},
# Bad data (should be rejected by quality gates)
{"customer_id": "", "customer_name": "", "segment": "Retail", "city": "Vilnius", "country_code": "LT"},
{"customer_id": "VIL-C010", "customer_name": "Test Customer DO NOT USE", "segment": "", "city": "", "country_code": "XX"},
]
filepath = os.path.join(os.path.dirname(__file__), "vilandra_customers.csv")
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["customer_id", "customer_name", "segment", "city", "country_code"])
writer.writeheader()
writer.writerows(customers)
print(f"Generated {len(customers)} customer records -> {filepath}")
return filepathWhat does this data represent?
The customer list is designed to exercise every path through the matching pipeline. The first 3 records are near-exact matches to existing BalticDW customers (they will auto-match). Records 4-5 are borderline similar (they will be quarantined for human review). Records 6-8 are genuinely new. The last 2 are intentionally bad data that should be rejected.
- Add the product generation function and the main entry point below:
def generate_products():
"""Create Vilandra product CSV with overlapping and new product entries."""
products = [
# Overlaps with existing BalticDW products
{"product_id": "VIL-P001", "product_name": "Colgate Total Toothpaste", "category": "Oral Care", "brand": "Colgate"},
{"product_id": "VIL-P002", "product_name": "Tork Premium Hand Towel", "category": "Tissue", "brand": "Tork"},
{"product_id": "VIL-P003", "product_name": "Palmolive Shower Gel 250ml", "category": "Personal Care", "brand": "Palmolive"},
# Borderline matches
{"product_id": "VIL-P004", "product_name": "Castrol Edge Motor Oil 5W30", "category": "Lubricants", "brand": "Castrol"},
# Genuinely new products
{"product_id": "VIL-P005", "product_name": "Svyturys Ekstra Lager 500ml", "category": "Beverages", "brand": "Svyturys"},
{"product_id": "VIL-P006", "product_name": "Zalgiris Energy Drink 330ml", "category": "Beverages", "brand": "Zalgiris"},
# Bad data
{"product_id": "", "product_name": "", "category": "", "brand": ""},
]
filepath = os.path.join(os.path.dirname(__file__), "vilandra_products.csv")
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["product_id", "product_name", "category", "brand"])
writer.writeheader()
writer.writerows(products)
print(f"Generated {len(products)} product records -> {filepath}")
return filepath
if __name__ == "__main__":
generate_customers()
generate_products()
print("\nVilandra acquisition data generated successfully.")What does this code do?
The product data follows the same pattern: 3 near-exact matches to existing BalticDW products, 1 borderline match, 2 genuinely new products, and 1 garbage row. The if __name__ block calls both generators when you run the script directly.
- Save generate_vilandra.py.
- Before you run this, what do you expect to see in the terminal output?
- Run the script:
python generate_vilandra.pyYou should see output confirming both files were generated:
Generated 10 customer records -> ...\vilandra_customers.csv
Generated 7 product records -> ...\vilandra_products.csv
Vilandra acquisition data generated successfully.- Confirm the CSV files exist by running:
ls vilandra_*.csvYou should see vilandra_customers.csv and vilandra_products.csv listed.
Seeing a FileNotFoundError or no CSV output?
Make sure you are running the script from the same directory where generate_vilandra.py is saved. The script writes CSV files relative to its own location using os.path.dirname(__file__).
Help me fix this CSV generation issue
✔️ Awesome, I've got everything!
Great. Double check you have saved both setup_mdm_schema.sql and generate_vilandra.py. Your MDM infrastructure is ready.
ⓧ I'd like to double check the full code
Scroll up and confirm your files contain all of the following sections in order:
- setup_mdm_schema.sql: Schema creation + staging_customers table, then staging_products table, then quarantine table, then merge_log table + PRINT statement.
- generate_vilandra.py: Module docstring + imports, then generate_customers() function (10 customer records), then generate_products() function (7 product records), then if __name__ block calling both.
Your MDM staging area is live and loaded with test data. Next up, you will see exactly why this infrastructure is necessary by attempting a naive load without it.
Attempt the Naive Load
Your MDM environment is ready with staging tables, quarantine infrastructure, and Vilandra's data waiting in CSV files.
The simplest approach to onboarding acquisition data would be to read the CSV and insert rows directly into your production dimensions. Let's try that and see what happens.
In this step, get ready to:
- Load Vilandra customer data directly into DimCustomer without any matching.
- Run dbt test and observe what breaks.
- Roll back the naive insert to restore DimCustomer to its clean state.
Load Vilandra data directly into DimCustomer
You have 8 valid Vilandra customer records (after filtering out empty rows) in your CSV. If you insert them straight into dbo.DimCustomer without checking whether they already exist, some will duplicate customers that are already in your warehouse.
- Create a new file called naive_load.py in your project directory.
- Add the file header, imports, and the first half of the naive_insert function by pasting this code:
"""Attempt a naive load of Vilandra data directly into DimCustomer.
This WILL create duplicates and break dbt tests. That is the point."""
import pyodbc
import pandas as pd
def naive_insert():
"""Insert Vilandra customers directly into DimCustomer without matching."""
conn = pyodbc.connect(
"Driver={ODBC Driver 18 for SQL Server};"
"Server=localhost;"
"Database=BalticDW;"
"Trusted_Connection=yes;"
"TrustServerCertificate=yes;"
)
cursor = conn.cursor()
df = pd.read_csv("vilandra_customers.csv")
# Filter out obviously empty rows for this naive attempt
df = df[df["customer_name"].notna() & (df["customer_name"] != "")]
print(f"Naive load: inserting {len(df)} Vilandra customers directly into DimCustomer...")
# Get the next available surrogate key
cursor.execute("SELECT ISNULL(MAX(customer_key), 0) FROM dbo.DimCustomer")
max_key = cursor.fetchone()[0]What does this code do?
- Connects to BalticDW using the same connection string from your earlier projects.
- Reads vilandra_customers.csv into a DataFrame and drops rows with empty names.
- Finds the current highest surrogate key in DimCustomer so new rows get unique keys.
- Directly below the code you just pasted, add the insert loop that writes each row into the production table:
for i, row in df.iterrows():
max_key += 1
cursor.execute(
"""
INSERT INTO dbo.DimCustomer (customer_key, customer_id, customer_name, segment, city, country_code, is_current)
VALUES (?, ?, ?, ?, ?, ?, 1)
""",
max_key, row["customer_id"], row["customer_name"],
row["segment"], row["city"], row["country_code"]
)
conn.commit()
print(f"Inserted {len(df)} rows. Now run 'dbt test' to see what breaks.")
cursor.close()
conn.close()What does this code do?
The loop iterates over every valid row and INSERTs it directly into dbo.DimCustomer with a new surrogate key. There is no check for whether a matching customer already exists.
This is the "naive" approach. Names like "Baltic Fuel SpA" will land alongside the existing "Baltic Fuel" record, creating a duplicate entity in your dimension.
- Finally, add the rollback function and the main block below naive_insert:
def rollback_naive():
"""Remove Vilandra rows inserted by the naive load."""
conn = pyodbc.connect(
"Driver={ODBC Driver 18 for SQL Server};"
"Server=localhost;"
"Database=BalticDW;"
"Trusted_Connection=yes;"
"TrustServerCertificate=yes;"
)
cursor = conn.cursor()
cursor.execute("DELETE FROM dbo.DimCustomer WHERE customer_id LIKE 'VIL-%'")
deleted = cursor.rowcount
conn.commit()
print(f"Rolled back {deleted} naive-load rows from DimCustomer.")
cursor.close()
conn.close()
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--rollback":
rollback_naive()
else:
naive_insert()What does this code do?
The rollback_naive function deletes any row where customer_id starts with VIL-. This is the safety net that lets you undo the naive insert cleanly.
The main block dispatches between insert and rollback based on a command-line flag. Running with --rollback triggers the cleanup.
- Save naive_load.py.
- Run the naive insert by executing this command in your terminal:
python naive_load.pyYou should see output confirming that 8 or 9 rows were inserted directly into DimCustomer.
Seeing a connection error?
- Make sure SQL Server is running. Check Services (SQL Server (MSSQLSERVER) should show Running).
- Confirm vilandra_customers.csv is in the same directory as naive_load.py.
help me debug a pyodbc connection error to SQL Server
✔️ Awesome, I've got everything!
Great. Double check you have saved naive_load.py and confirmed the insert succeeded.
ⓧ I'd like to double check the full code
Scroll up and confirm your naive_load.py contains these sections in order:
- File docstring and imports (pyodbc, pandas).
- The naive_insert() function: connection, CSV read, filter, surrogate key lookup, insert loop, commit, and close.
- The rollback_naive() function: connection, DELETE WHERE customer_id LIKE 'VIL-%', commit, and close.
- The if __name__ == "__main__" block dispatching between insert and --rollback.
Run dbt test to see the failure
Your dbt project from Project 2 has tests that validate data integrity in your staging models. Those tests are about to catch the mess you just created.
Before you run the test, think about this: you just inserted rows like "Baltic Fuel SpA" alongside an existing "Baltic Fuel" record. Do you think dbt's tests will pass or fail?
- Switch to your dbt project directory and run the following command:
dbt test --select stg_sales_ordersYou will see at least one test FAIL with a message about duplicate or invalid records.
Why did this fail?
The naive insert created duplicate natural keys. Customers like "Baltic Fuel SpA" now coexist with "Baltic Fuel" as separate dimension rows, even though they represent the same real-world entity.
Your dbt tests enforce assumptions about data quality (uniqueness, referential integrity) that the raw insert violated. This is exactly why MDM pipelines exist.
Tests passed unexpectedly?
If all tests pass, confirm that naive_load.py actually ran successfully in the previous substep. Check dbo.DimCustomer for rows where customer_id LIKE 'VIL-%'. If none exist, the insert did not complete.
help me figure out why dbt test passed when I expected it to fail after a naive insert
Roll back the naive insert
You have proven that raw acquisition data cannot go straight into production dimensions. Now clean up the damage so your warehouse is back to its Project 3 state.
- Return to your project directory and run the rollback by executing:
python naive_load.py --rollbackYou should see output confirming that the Vilandra rows were deleted (e.g. "Rolled back 8 naive-load rows from DimCustomer.").
- Verify the cleanup by running dbt test --select stg_sales_orders again. All tests should pass now that the duplicates are gone.
Rollback didn't delete any rows?
The rollback targets rows where customer_id LIKE 'VIL-%'. If you see "Rolled back 0 rows", the naive insert may have failed earlier or you already ran the rollback once.
help me verify that DimCustomer is back to its clean state after the naive load rollback
Your BalticDW is clean again. The naive approach failed because it treated acquisition data as pre-matched truth. Next up, you will build an entity resolution pipeline that uses fuzzy matching to determine whether an incoming record is an existing entity, a new one, or something ambiguous that needs human review.
Build the Entity Resolution Pipeline
In Step 2, you watched the naive insert create duplicates that broke your dbt tests. You rolled it back and now DimCustomer is clean again.
The core problem is that you treated acquisition data as pre-matched truth. Entity resolution solves this by using fuzzy string matching to determine whether each incoming record represents an existing entity or a genuinely new one. Each record gets classified into one of three buckets: auto-match, quarantine, or reject.
In this step, get ready to:
- Load Vilandra data into the mdm staging tables as the proper landing zone.
- Build the fuzzy matching function using RapidFuzz's three-way threshold classification.
- Run entity resolution and review the classification results for every Vilandra record.
Load Vilandra data into the mdm staging tables
Before matching begins, acquisition data must land in the mdm staging schema you created in Step 1. This keeps it isolated from production until entity resolution decides what belongs.
- Load both Vilandra CSVs into the staging tables by running this Python snippet in your terminal:
import pyodbc, pandas as pd
conn = pyodbc.connect(
"Driver={ODBC Driver 18 for SQL Server};"
"Server=localhost;Database=BalticDW;"
"Trusted_Connection=yes;TrustServerCertificate=yes;"
)
cursor = conn.cursor()
# Load customers into staging
cust_df = pd.read_csv("vilandra_customers.csv")
for _, row in cust_df.iterrows():
cursor.execute(
"INSERT INTO mdm.staging_customers (source_system, customer_id, customer_name, segment, city, country_code) VALUES ('UAB_VILANDRA', ?, ?, ?, ?, ?)",
row["customer_id"], row["customer_name"], row["segment"], row["city"], row["country_code"]
)
# Load products into staging
prod_df = pd.read_csv("vilandra_products.csv")
for _, row in prod_df.iterrows():
cursor.execute(
"INSERT INTO mdm.staging_products (source_system, product_id, product_name, category, brand) VALUES ('UAB_VILANDRA', ?, ?, ?, ?)",
row["product_id"], row["product_name"], row["category"], row["brand"]
)
conn.commit()
print(f"Loaded {len(cust_df)} customers and {len(prod_df)} products into staging.")
cursor.close()
conn.close()What does this code do?
This reads both Vilandra CSVs and inserts every row into mdm.staging_customers and mdm.staging_products with a source_system tag of UAB_VILANDRA.
The staging tables are your controlled landing zone. Nothing here touches DimCustomer or DimProduct until entity resolution says it is safe.
- Run the script:
python load_staging.py- Verify the data landed by running this quick count query in your terminal:
python -c "import pyodbc; conn = pyodbc.connect('Driver={ODBC Driver 18 for SQL Server};Server=localhost;Database=BalticDW;Trusted_Connection=yes;TrustServerCertificate=yes;'); cursor = conn.cursor(); cursor.execute('SELECT COUNT(*) FROM mdm.staging_customers'); print(f'Staging customers: {cursor.fetchone()[0]}'); cursor.execute('SELECT COUNT(*) FROM mdm.staging_products'); print(f'Staging products: {cursor.fetchone()[0]}'); cursor.close(); conn.close()"You should see Staging customers: 10 and Staging products: 7.
Seeing 0 rows?
Make sure you are running from the same directory where vilandra_customers.csv and vilandra_products.csv live. The script reads CSVs from the current working directory.
If you see a FileNotFoundError, re-run generate_vilandra.py from Step 1 to regenerate the CSV files.
Help me debug my staging table load.
Build the fuzzy matching function
Now for the core matching logic. You will create entity_resolution.py which uses RapidFuzz's process.extractOne() to compare each incoming name against your canonical dimensions and classify it.
- Create a new file called entity_resolution.py in your project directory.
- Add the imports, matching thresholds, and connection helper by pasting this code:
"""Entity resolution pipeline using RapidFuzz for fuzzy matching."""
import pyodbc
import pandas as pd
from rapidfuzz import fuzz, process, utils
# Thresholds for three-way classification
AUTO_MATCH_THRESHOLD = 90
QUARANTINE_THRESHOLD = 70
def get_connection():
"""Create a pyodbc connection to BalticDW."""
return pyodbc.connect(
"Driver={ODBC Driver 18 for SQL Server};"
"Server=localhost;"
"Database=BalticDW;"
"Trusted_Connection=yes;"
"TrustServerCertificate=yes;"
)What do the thresholds mean?
AUTO_MATCH_THRESHOLD = 90 means any record scoring 90 or above is confident enough to merge automatically.
QUARANTINE_THRESHOLD = 70 defines the lower bound. Records scoring 70-89 are too similar to ignore but too uncertain to auto-merge. They go to quarantine for human review. Below 70, the record either has no match (new entity) or is rejected by validation.
- Below get_connection(), add the functions that load your canonical dimensions and validate incoming records:
def load_canonical_customers(conn):
"""Load existing canonical customers from DimCustomer."""
query = """
SELECT customer_key, customer_id, customer_name, segment, city, country_code
FROM dbo.DimCustomer
WHERE is_current = 1
"""
return pd.read_sql(query, conn)
def load_canonical_products(conn):
"""Load existing canonical products from DimProduct."""
query = """
SELECT product_key, product_id, product_name, category, brand
FROM dbo.DimProduct
"""
return pd.read_sql(query, conn)
def validate_record(record, entity_type):
"""Data quality gate: reject records that fail basic validation."""
if entity_type == "customer":
if not record.get("customer_id") or not record.get("customer_name"):
return False, "MISSING_REQUIRED_FIELD"
if record.get("country_code") and len(record["country_code"]) != 2:
return False, "INVALID_COUNTRY_CODE"
elif entity_type == "product":
if not record.get("product_id") or not record.get("product_name"):
return False, "MISSING_REQUIRED_FIELD"
return True, NoneWhat does validate_record do?
This is your first data quality gate. Before any fuzzy matching happens, each record is checked for required fields and valid formats.
Records with empty IDs, empty names, or invalid country codes get immediately rejected. They never reach the matching engine. Remember those bad-data rows in your Vilandra CSV? This is what catches them.
- Save entity_resolution.py.
- Verify the imports load correctly by running:
python -c "from entity_resolution import validate_record; print(validate_record({'customer_id': '', 'customer_name': ''}, 'customer'))"You should see (False, 'MISSING_REQUIRED_FIELD') confirming the validator rejects empty records.
Now for the heart of entity resolution: the function that scores every incoming customer against your canonical dimension.
- Below validate_record, add the first part of resolve_customers which sets up the loop, runs validation, and calls RapidFuzz:
def resolve_customers(staging_df, canonical_df, conn):
"""Resolve customer entities using fuzzy matching."""
canonical_names = canonical_df["customer_name"].tolist()
canonical_keys = canonical_df["customer_key"].tolist()
results = {"auto_match": [], "quarantine": [], "new_entity": [], "rejected": []}
for _, row in staging_df.iterrows():
record = row.to_dict()
# Data quality gate
is_valid, reject_reason = validate_record(record, "customer")
if not is_valid:
results["rejected"].append({**record, "reason": reject_reason})
print(f" REJECTED: {record.get('customer_id', '?')} - {reject_reason}")
continue
# Fuzzy match against canonical names
match = process.extractOne(
record["customer_name"],
canonical_names,
scorer=fuzz.token_sort_ratio,
processor=utils.default_process,
score_cutoff=QUARANTINE_THRESHOLD
)What does process.extractOne do?
process.extractOne() takes an incoming name and finds the single best match from your list of canonical names. It returns a tuple of (matched_name, score, index) or None if nothing scores above the cutoff.
scorer=fuzz.token_sort_ratio normalizes word order before comparing. This means "Baltic Fuel SpA" matches "Baltic Fuel" even though the words are in different positions.
processor=utils.default_process strips non-alphanumeric characters and lowercases both strings before scoring. Since RapidFuzz 3.0.0+, this preprocessing is not applied by default.
- Directly below the process.extractOne() call (still inside the for loop), add the classification logic that routes each match into the correct bucket:
if match is None:
# No match above quarantine threshold - genuinely new entity
results["new_entity"].append(record)
print(f" NEW ENTITY: {record['customer_name']} (no match >= {QUARANTINE_THRESHOLD})")
else:
matched_name, score, idx = match
matched_key = canonical_keys[idx]
if score >= AUTO_MATCH_THRESHOLD:
results["auto_match"].append({
**record,
"matched_key": matched_key,
"matched_name": matched_name,
"score": score
})
print(f" AUTO-MATCH: {record['customer_name']} -> {matched_name} (score: {score:.0f})")
else:
results["quarantine"].append({
**record,
"matched_key": matched_key,
"matched_name": matched_name,
"score": score
})
print(f" QUARANTINE: {record['customer_name']} ~> {matched_name} (score: {score:.0f})")
return resultsWhat does the three-way classification do?
If match is None: no canonical record scored above 70. This is a genuinely new entity that should be inserted fresh.
If the score is 90 or above: the match is confident enough for automatic merge. "Baltic Fuel SpA" scoring 92 against "Baltic Fuel" is clearly the same company.
If the score is 70-89: the match is ambiguous. It goes to quarantine for a human data steward to review later.
- Save entity_resolution.py.
The product resolver follows the exact same pattern but matches against DimProduct names instead.
- Below resolve_customers, add the resolve_products function by pasting this code:
def resolve_products(staging_df, canonical_df, conn):
"""Resolve product entities using fuzzy matching."""
canonical_names = canonical_df["product_name"].tolist()
canonical_keys = canonical_df["product_key"].tolist()
results = {"auto_match": [], "quarantine": [], "new_entity": [], "rejected": []}
for _, row in staging_df.iterrows():
record = row.to_dict()
is_valid, reject_reason = validate_record(record, "product")
if not is_valid:
results["rejected"].append({**record, "reason": reject_reason})
print(f" REJECTED: {record.get('product_id', '?')} - {reject_reason}")
continue
match = process.extractOne(
record["product_name"],
canonical_names,
scorer=fuzz.token_sort_ratio,
processor=utils.default_process,
score_cutoff=QUARANTINE_THRESHOLD
)- Directly below the process.extractOne() call in resolve_products, add the same classification logic for products:
if match is None:
results["new_entity"].append(record)
print(f" NEW ENTITY: {record['product_name']} (no match >= {QUARANTINE_THRESHOLD})")
else:
matched_name, score, idx = match
matched_key = canonical_keys[idx]
if score >= AUTO_MATCH_THRESHOLD:
results["auto_match"].append({
**record,
"matched_key": matched_key,
"matched_name": matched_name,
"score": score
})
print(f" AUTO-MATCH: {record['product_name']} -> {matched_name} (score: {score:.0f})")
else:
results["quarantine"].append({
**record,
"matched_key": matched_key,
"matched_name": matched_name,
"score": score
})
print(f" QUARANTINE: {record['product_name']} ~> {matched_name} (score: {score:.0f})")
return results- Save entity_resolution.py and verify the resolve functions import correctly:
python -c "from entity_resolution import resolve_customers, resolve_products; print('Resolve functions loaded OK')"You should see Resolve functions loaded OK.
Seeing an IndentationError?
The classification logic (the if match is None block) must be indented inside the for loop at the same level as the process.extractOne() call above it. Check that both chunks are at 8 spaces of indentation (two levels deep).
The return results line should be at 4 spaces (one level), aligned with the for keyword.
Help me fix my indentation in entity_resolution.py.
Run entity resolution and review results
The matching functions are ready. Now you need the write_quarantine function that persists ambiguous records to your quarantine table, and a main() function that orchestrates the full pipeline.
- Below resolve_products, add the quarantine writer:
def write_quarantine(results, entity_type, conn):
"""Write quarantined records to mdm.quarantine table."""
cursor = conn.cursor()
for record in results["quarantine"]:
cursor.execute(
"""
INSERT INTO mdm.quarantine (source_system, entity_type, source_id, source_name,
candidate_key, candidate_name, match_score)
VALUES ('UAB_VILANDRA', ?, ?, ?, ?, ?, ?)
""",
entity_type,
record.get("customer_id") or record.get("product_id"),
record.get("customer_name") or record.get("product_name"),
record["matched_key"],
record["matched_name"],
record["score"]
)
conn.commit()
cursor.close()What does write_quarantine do?
For every record classified as quarantine (score 70-89), this function inserts a row into mdm.quarantine with the source name, the candidate match from your canonical dimension, and the similarity score.
These rows start with status = 'PENDING' (the table default). A human data steward can review them later.
- Below write_quarantine, add the first part of the main() function which loads data, runs resolution, and writes quarantine records:
def main():
"""Run the full entity resolution pipeline."""
conn = get_connection()
# Load staging data
customers_df = pd.read_csv("vilandra_customers.csv")
products_df = pd.read_csv("vilandra_products.csv")
# Load canonical dimensions
canonical_customers = load_canonical_customers(conn)
canonical_products = load_canonical_products(conn)
print(f"\nCanonical customers: {len(canonical_customers)}")
print(f"Canonical products: {len(canonical_products)}")
# Resolve customers
print(f"\n{'='*60}")
print("CUSTOMER ENTITY RESOLUTION")
print(f"{'='*60}")
print(f"Incoming records: {len(customers_df)}")
print(f"Thresholds: auto-match >= {AUTO_MATCH_THRESHOLD}, quarantine >= {QUARANTINE_THRESHOLD}\n")
customer_results = resolve_customers(customers_df, canonical_customers, conn)- Directly below the customer_results line, add the product resolution, summary output, and the entry point:
# Resolve products
print(f"\n{'='*60}")
print("PRODUCT ENTITY RESOLUTION")
print(f"{'='*60}")
print(f"Incoming records: {len(products_df)}")
print(f"Thresholds: auto-match >= {AUTO_MATCH_THRESHOLD}, quarantine >= {QUARANTINE_THRESHOLD}\n")
product_results = resolve_products(products_df, canonical_products, conn)
# Write quarantined records
write_quarantine(customer_results, "customer", conn)
write_quarantine(product_results, "product", conn)
# Summary
print(f"\n{'='*60}")
print("RESOLUTION SUMMARY")
print(f"{'='*60}")
print(f"\nCustomers:")
print(f" Auto-match: {len(customer_results['auto_match'])}")
print(f" Quarantine: {len(customer_results['quarantine'])}")
print(f" New entity: {len(customer_results['new_entity'])}")
print(f" Rejected: {len(customer_results['rejected'])}")- Finally, add the product summary, result save, and entry point at the bottom of main():
print(f"\nProducts:")
print(f" Auto-match: {len(product_results['auto_match'])}")
print(f" Quarantine: {len(product_results['quarantine'])}")
print(f" New entity: {len(product_results['new_entity'])}")
print(f" Rejected: {len(product_results['rejected'])}")
conn.close()
# Save results for the survivorship step
import json
results_payload = {
"customers": customer_results,
"products": product_results
}
with open("resolution_results.json", "w", encoding="utf-8") as f:
json.dump(results_payload, f, indent=2, default=str)
print(f"\nResults saved to resolution_results.json")
if __name__ == "__main__":
main()Why save results to JSON?
The resolution_results.json file passes the classification results to the next step (survivorship and merge). This decouples the resolution phase from the merge phase so you can re-run either independently.
- Save entity_resolution.py.
Before you run the pipeline, what do you expect to see? Think about the Vilandra customer names. "Baltic Fuel SpA" should score highly against "Baltic Fuel". "Vilandra Beverages OOO" has no match in your existing data. The empty rows should be rejected.
- Run the full entity resolution pipeline:
python entity_resolution.pyYou should see output showing each record classified with its match score. Look for lines like:
- AUTO-MATCH: Baltic Fuel SpA -> Baltic Fuel (score: 92) for confident matches.
- QUARANTINE: lines for borderline records (score 70-89).
- NEW ENTITY: lines for Vilandra-only names like "Vilandra Beverages OOO".
- REJECTED: lines for the empty and invalid rows.
At the bottom, the resolution summary shows the count in each category for both customers and products.
Why token_sort_ratio instead of simple ratio?
Company names frequently appear in different word orders across systems. "Baltic Fuel SpA" vs "SpA Baltic Fuel" would score poorly with a simple left-to-right comparison.
token_sort_ratio alphabetically sorts the words in both strings before comparing. This means word-order differences do not penalize the score. For entity names in MDM, this is the standard choice.
- Confirm quarantined records were written to the database by running:
python -c "import pyodbc; conn = pyodbc.connect('Driver={ODBC Driver 18 for SQL Server};Server=localhost;Database=BalticDW;Trusted_Connection=yes;TrustServerCertificate=yes;'); cursor = conn.cursor(); cursor.execute('SELECT source_name, candidate_name, match_score, status FROM mdm.quarantine'); rows = cursor.fetchall(); [print(f'{r[0]} ~> {r[1]} (score: {r[2]:.0f}, status: {r[3]})') for r in rows]; cursor.close(); conn.close()"You should see the borderline matches listed with their scores and status: PENDING.
Seeing no AUTO-MATCH results?
If all records show as NEW ENTITY, your DimCustomer table may be empty. Confirm your BalticDW dimensions have data by running: SELECT COUNT(*) FROM dbo.DimCustomer WHERE is_current = 1.
If you see an ImportError on rapidfuzz, confirm it is installed with pip show rapidfuzz.
Help me troubleshoot my entity resolution output.
✔️ Awesome, I've got everything!
Confirm you have saved entity_resolution.py and that resolution_results.json exists in your project directory.
ⓧ I'd like to double check the full code
Your entity_resolution.py should contain these sections in order. Scroll up and confirm each is present:
- Imports (pyodbc, pandas, rapidfuzz) and threshold constants (AUTO_MATCH_THRESHOLD = 90, QUARANTINE_THRESHOLD = 70).
- get_connection() returning a pyodbc connection to BalticDW.
- load_canonical_customers(conn) and load_canonical_products(conn) querying the production dimensions.
- validate_record(record, entity_type) checking for missing fields and invalid country codes.
- resolve_customers(staging_df, canonical_df, conn) with the full loop, extractOne call, and three-way classification.
- resolve_products(staging_df, canonical_df, conn) following the same pattern.
- write_quarantine(results, entity_type, conn) inserting into mdm.quarantine.
- main() orchestrating the pipeline, printing the summary, and saving resolution_results.json.
- if __name__ == "__main__": main() at the bottom.
Your entity resolution pipeline is classifying every Vilandra record with a confidence score. Next up, you will apply survivorship rules to decide exactly which attributes to keep from each match and merge the winners into your production dimensions.
Apply Survivorship Rules and Merge
Your entity resolution pipeline classified every Vilandra record. You know WHICH records match existing entities in BalticDW.
Now you need to decide WHAT to keep from each match. Survivorship rules define per-attribute which source wins, creating a single golden record from multiple sources without losing the audit trail.
In this step, get ready to:
- Define per-attribute survivorship logic (source priority, completeness, recency).
- Merge resolved records into production dimensions with a full audit log.
- Insert Vilandra's subsidiary and fact data with correctly mapped surrogate keys.
Define survivorship rules
When two sources describe the same entity, you need a deterministic rule for each attribute. Your BalticDW has been production-validated across three projects, so it is the trusted source. Vilandra's data is the newcomer.
- Create a new file called survivorship_merge.py in your project folder.
- Add the module header with imports and the survivorship rules configuration:
"""Apply survivorship rules and merge resolved records into production dimensions."""
import json
import pyodbc
import pandas as pd
from datetime import datetime
# Survivorship rules per attribute
SURVIVORSHIP_RULES = {
"customer_name": "source_priority", # Existing BalticDW wins for canonical names
"segment": "completeness", # Non-null wins over null
"city": "completeness", # Non-null wins over null
"country_code": "source_priority", # Existing wins
"product_name": "source_priority", # Existing wins
"category": "source_priority", # Existing wins
"brand": "completeness", # Non-null wins over null
}What do these rules mean?
Each attribute maps to a strategy that determines which value survives into the golden record.
- source_priority: The existing BalticDW value wins. It has been validated in production.
- completeness: Whichever value is non-null wins. This fills gaps without overwriting.
- recency: The incoming (newer) value wins. Used for contact details that change over time.
- Below the rules dictionary, add the connection helper and the survivorship decision function:
def get_connection():
"""Create a pyodbc connection to BalticDW."""
return pyodbc.connect(
"Driver={ODBC Driver 18 for SQL Server};"
"Server=localhost;"
"Database=BalticDW;"
"Trusted_Connection=yes;"
"TrustServerCertificate=yes;"
)
def apply_survivorship(existing_value, incoming_value, rule):
"""Apply a survivorship rule to determine the winning value."""
if rule == "source_priority":
# Existing (BalticDW) wins unless it is null/empty
if existing_value and str(existing_value).strip():
return existing_value, "existing_wins"
return incoming_value, "incoming_fills_gap"
elif rule == "completeness":
# Non-null/non-empty wins
if incoming_value and str(incoming_value).strip():
if not existing_value or not str(existing_value).strip():
return incoming_value, "incoming_fills_gap"
return existing_value, "existing_wins"
elif rule == "recency":
# Incoming (newer) wins
return incoming_value, "incoming_wins_recency"
return existing_value, "default_existing"What does apply_survivorship return?
It returns a tuple: the winning value and a string describing the decision. The decision string becomes the audit trail in your merge log.
- Add the audit logging function below apply_survivorship:
def log_merge_decision(cursor, entity_type, entity_key, source_id, attr_name,
old_value, new_value, rule, steward_override=False):
"""Write a merge decision to the audit log."""
cursor.execute(
"""
INSERT INTO mdm.merge_log (entity_type, entity_key, source_system, source_id,
attribute_name, old_value, new_value, survivorship_rule, steward_override)
VALUES (?, ?, 'UAB_VILANDRA', ?, ?, ?, ?, ?, ?)
""",
entity_type, entity_key, source_id, attr_name,
str(old_value) if old_value else None,
str(new_value) if new_value else None,
rule, steward_override
)Why log every single attribute decision?
This is the audit trail that MDM demands. Every attribute-level decision is recorded with what won, why, and when. If anyone asks "why did Baltic Fuel's city change?", the merge log gives a definitive answer.
- Save survivorship_merge.py.
- Verify the module loads by running this command in your terminal:
python -c "from survivorship_merge import apply_survivorship, SURVIVORSHIP_RULES; print(SURVIVORSHIP_RULES)"You should see the rules dictionary printed back, confirming the module imports cleanly.
Seeing an ImportError?
- Make sure your terminal is in the same directory as survivorship_merge.py.
- Check for a typo in the function name or missing colon at the end of the def line.
- help me debug an ImportError in survivorship_merge.py
Execute the merge into production dimensions
With the survivorship logic defined, you can now write the functions that apply it to each auto-matched record and insert genuinely new entities.
- In survivorship_merge.py, add the first part of the customer merge function below log_merge_decision:
def merge_auto_matched_customers(results, conn):
"""Merge auto-matched customers into DimCustomer using survivorship rules."""
cursor = conn.cursor()
merged_count = 0
for record in results["auto_match"]:
matched_key = record["matched_key"]
source_id = record["customer_id"]
# Fetch current canonical record
cursor.execute(
"SELECT customer_name, segment, city, country_code FROM dbo.DimCustomer WHERE customer_key = ?",
matched_key
)
existing = cursor.fetchone()
if not existing:
continue
existing_values = {
"customer_name": existing[0],
"segment": existing[1],
"city": existing[2],
"country_code": existing[3]
}- Continue the same function (still inside the for loop) with the survivorship decision and update logic:
# Apply survivorship for each attribute
updates = {}
for attr in ["segment", "city"]:
rule = SURVIVORSHIP_RULES.get(attr, "source_priority")
winner, decision = apply_survivorship(
existing_values[attr], record.get(attr), rule
)
if winner != existing_values[attr]:
updates[attr] = winner
log_merge_decision(cursor, "customer", matched_key, source_id,
attr, existing_values[attr], winner, decision)
else:
log_merge_decision(cursor, "customer", matched_key, source_id,
attr, existing_values[attr], existing_values[attr], decision)
# Apply updates if any attributes changed
if updates:
set_clause = ", ".join(f"{k} = ?" for k in updates.keys())
cursor.execute(
f"UPDATE dbo.DimCustomer SET {set_clause} WHERE customer_key = ?",
*updates.values(), matched_key
)
merged_count += 1
conn.commit()
cursor.close()
print(f" Merged {merged_count} auto-matched customers (survivorship applied)")
return merged_countWhat does this function do?
- For each auto-matched customer, it fetches the current canonical record from DimCustomer.
- It applies survivorship rules to segment and city (completeness rule). Names and country codes use source_priority, so existing values always win.
- Every decision is logged to mdm.merge_log regardless of whether a value changed.
- Add the function that inserts genuinely new customers (entities with no match in BalticDW):
def insert_new_customers(results, conn):
"""Insert genuinely new customers into DimCustomer."""
cursor = conn.cursor()
cursor.execute("SELECT ISNULL(MAX(customer_key), 0) FROM dbo.DimCustomer")
max_key = cursor.fetchone()[0]
inserted = 0
for record in results["new_entity"]:
max_key += 1
cursor.execute(
"""
INSERT INTO dbo.DimCustomer (customer_key, customer_id, customer_name, segment, city, country_code, is_current)
VALUES (?, ?, ?, ?, ?, ?, 1)
""",
max_key, record["customer_id"], record["customer_name"],
record["segment"], record["city"], record["country_code"]
)
log_merge_decision(cursor, "customer", max_key, record["customer_id"],
"NEW_ENTITY", None, record["customer_name"], "new_insert")
inserted += 1
conn.commit()
cursor.close()
print(f" Inserted {inserted} new customer entities")
return insertedWhy log NEW_ENTITY as an attribute?
Even brand-new inserts get a merge log entry. This makes the audit trail complete. You can query the log later to see every entity that entered the warehouse and exactly how.
- Add the product merge function. It follows the same pattern but applies survivorship only to the brand attribute:
def merge_auto_matched_products(results, conn):
"""Merge auto-matched products into DimProduct using survivorship rules."""
cursor = conn.cursor()
merged_count = 0
for record in results["auto_match"]:
matched_key = record["matched_key"]
source_id = record["product_id"]
cursor.execute(
"SELECT product_name, category, brand FROM dbo.DimProduct WHERE product_key = ?",
matched_key
)
existing = cursor.fetchone()
if not existing:
continue
existing_values = {
"product_name": existing[0],
"category": existing[1],
"brand": existing[2]
}- Continue merge_auto_matched_products with the survivorship and commit logic:
updates = {}
for attr in ["brand"]:
rule = SURVIVORSHIP_RULES.get(attr, "source_priority")
winner, decision = apply_survivorship(
existing_values[attr], record.get(attr), rule
)
if winner != existing_values[attr]:
updates[attr] = winner
log_merge_decision(cursor, "product", matched_key, source_id,
attr, existing_values[attr], winner, decision)
else:
log_merge_decision(cursor, "product", matched_key, source_id,
attr, existing_values[attr], existing_values[attr], decision)
if updates:
set_clause = ", ".join(f"{k} = ?" for k in updates.keys())
cursor.execute(
f"UPDATE dbo.DimProduct SET {set_clause} WHERE product_key = ?",
*updates.values(), matched_key
)
merged_count += 1
conn.commit()
cursor.close()
print(f" Merged {merged_count} auto-matched products (survivorship applied)")
return merged_count- Add the new product insertion function:
def insert_new_products(results, conn):
"""Insert genuinely new products into DimProduct."""
cursor = conn.cursor()
cursor.execute("SELECT ISNULL(MAX(product_key), 0) FROM dbo.DimProduct")
max_key = cursor.fetchone()[0]
inserted = 0
for record in results["new_entity"]:
max_key += 1
cursor.execute(
"""
INSERT INTO dbo.DimProduct (product_key, product_id, product_name, category, brand)
VALUES (?, ?, ?, ?, ?)
""",
max_key, record["product_id"], record["product_name"],
record["category"], record["brand"]
)
log_merge_decision(cursor, "product", max_key, record["product_id"],
"NEW_ENTITY", None, record["product_name"], "new_insert")
inserted += 1
conn.commit()
cursor.close()
print(f" Inserted {inserted} new product entities")
return inserted- Save survivorship_merge.py.
Insert Vilandra's fact data
Now that dimensions are resolved, you can safely insert Vilandra's sales orders. First you need a subsidiary row so the fact table has a valid foreign key.
- Add the subsidiary insertion function below insert_new_products:
def insert_vilandra_subsidiary(conn):
"""Add UAB Vilandra as a new subsidiary."""
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM dbo.DimSubsidiary WHERE subsidiary_name = 'UAB Vilandra'")
if cursor.fetchone()[0] > 0:
print(" UAB Vilandra subsidiary already exists, skipping.")
cursor.close()
return
cursor.execute("SELECT ISNULL(MAX(subsidiary_key), 0) FROM dbo.DimSubsidiary")
max_key = cursor.fetchone()[0] + 1
cursor.execute(
"""
INSERT INTO dbo.DimSubsidiary (subsidiary_key, country_code, country_name, subsidiary_name)
VALUES (?, 'LT', 'Lithuania', 'UAB Vilandra')
""",
max_key
)
conn.commit()
cursor.close()
print(f" Added UAB Vilandra subsidiary (key: {max_key})")- Add the first part of the fact data insertion function:
def insert_vilandra_facts(conn):
"""Insert sample Vilandra sales orders into FactSalesOrder."""
cursor = conn.cursor()
# Get Vilandra subsidiary key
cursor.execute("SELECT subsidiary_key FROM dbo.DimSubsidiary WHERE subsidiary_name = 'UAB Vilandra'")
row = cursor.fetchone()
if not row:
print(" ERROR: UAB Vilandra subsidiary not found. Run insert_vilandra_subsidiary first.")
cursor.close()
return
subsidiary_key = row[0]
# Get some customer and product keys for fact rows
cursor.execute("""
SELECT TOP 3 customer_key FROM dbo.DimCustomer
WHERE country_code = 'LT' AND is_current = 1
""")
customer_keys = [r[0] for r in cursor.fetchall()]
cursor.execute("SELECT TOP 3 product_key FROM dbo.DimProduct")
product_keys = [r[0] for r in cursor.fetchall()]
# Get a date key
cursor.execute("SELECT TOP 1 date_key FROM dbo.DimDate WHERE year = 2025 AND month = 6")
date_row = cursor.fetchone()
date_key = date_row[0] if date_row else 20250601- Continue insert_vilandra_facts with the order creation and commit:
# Insert sample fact rows
sample_orders = [
(date_key, customer_keys[0] if customer_keys else 1, product_keys[0] if product_keys else 1, subsidiary_key, 5, 12.50, 62.50),
(date_key, customer_keys[1] if len(customer_keys) > 1 else 1, product_keys[1] if len(product_keys) > 1 else 1, subsidiary_key, 10, 8.75, 87.50),
(date_key, customer_keys[2] if len(customer_keys) > 2 else 1, product_keys[2] if len(product_keys) > 2 else 1, subsidiary_key, 3, 45.00, 135.00),
]
cursor.execute("SELECT ISNULL(MAX(order_key), 0) FROM dbo.FactSalesOrder")
max_key = cursor.fetchone()[0]
for order in sample_orders:
max_key += 1
cursor.execute(
"""
INSERT INTO dbo.FactSalesOrder (order_key, date_key, customer_key, product_key, subsidiary_key, quantity, unit_price, line_total)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
max_key, *order
)
conn.commit()
cursor.close()
print(f" Inserted {len(sample_orders)} Vilandra sales order rows into FactSalesOrder")Why only 3 fact rows?
This project focuses on MDM, not bulk data loading. Three rows are enough to prove the surrogate key mapping works end-to-end. In production, the full Vilandra export would load thousands of rows using the same pattern.
- Finally, add the main() function that orchestrates the entire merge:
def main():
"""Run survivorship rules and merge resolved entities."""
# Load resolution results from previous step
with open("resolution_results.json", "r", encoding="utf-8") as f:
results = json.load(f)
conn = get_connection()
print(f"\n{'='*60}")
print("SURVIVORSHIP AND MERGE")
print(f"{'='*60}")
print("\nCustomer merge:")
merge_auto_matched_customers(results["customers"], conn)
insert_new_customers(results["customers"], conn)
print("\nProduct merge:")
merge_auto_matched_products(results["products"], conn)
insert_new_products(results["products"], conn)
print("\nSubsidiary setup:")
insert_vilandra_subsidiary(conn)
print("\nFact data:")
insert_vilandra_facts(conn)- Continue main() with the audit summary and the entry point:
# Print audit summary
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM mdm.merge_log")
log_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM mdm.quarantine WHERE status = 'PENDING'")
pending_count = cursor.fetchone()[0]
cursor.close()
print(f"\n{'='*60}")
print("MERGE COMPLETE")
print(f"{'='*60}")
print(f" Audit log entries: {log_count}")
print(f" Quarantined (pending review): {pending_count}")
conn.close()
if __name__ == "__main__":
main()- Save survivorship_merge.py.
Before you run this, what do you expect to see? Think about how many auto-matched customers and products the resolution step found, and how many new entities it identified.
- Run the survivorship merge pipeline:
python survivorship_merge.pyYou should see a structured report showing customers merged, new entities inserted, the UAB Vilandra subsidiary added, fact rows created, and a final count of audit log entries.
Seeing a KeyError or missing column?
- Make sure resolution_results.json exists in the same directory. It was created by entity_resolution.py in the previous step.
- Verify your DimCustomer table still has the is_current column from Project 1.
- help me debug a KeyError in survivorship_merge.py
- Verify the audit trail by querying mdm.merge_log in SSMS or VS Code:
SELECT * FROM mdm.merge_log ORDER BY log_id;You should see one row per merged attribute showing the entity type, entity key, source that won, the old value, the new value, and the survivorship rule that decided it.
What should the merge log look like?
Each row represents a single attribute-level decision. For auto-matched customers, you will see rows for segment and city with survivorship_rule values like existing_wins or incoming_fills_gap. For new entities, you will see NEW_ENTITY as the attribute name with rule new_insert.
✔️ Awesome, I've got everything!
Great. Double check you have saved survivorship_merge.py and confirmed the merge log output.
ⓧ I'd like to double check the full code
Scroll up and confirm your survivorship_merge.py contains all of the following sections in order:
- Module docstring, imports (json, pyodbc, pandas, datetime), and SURVIVORSHIP_RULES dictionary.
- get_connection() function.
- apply_survivorship() function with source_priority, completeness, and recency branches.
- log_merge_decision() function.
- merge_auto_matched_customers() function.
- insert_new_customers() function.
- merge_auto_matched_products() function.
- insert_new_products() function.
- insert_vilandra_subsidiary() function.
- insert_vilandra_facts() function.
- main() function and if __name__ == "__main__" block.
Your production dimensions now contain properly merged Vilandra data with a complete audit trail. Next up, you will run dbt build to prove the entire platform still works end-to-end.
Run dbt build and Validate End-to-End
The merge is done. Your production dimensions now contain Vilandra's resolved data, and every decision is auditable in mdm.merge_log.
But you need to prove the entire platform still works end-to-end. Running dbt build executes both model transformations and tests in dependency order. If everything passes, Vilandra's data is flowing cleanly through your reporting layer.
In this step, get ready to:
- Run dbt build and confirm all models and tests pass.
- Query reporting marts to verify Vilandra revenue appears in group totals.
- Refresh Power BI and confirm the new subsidiary shows in your visuals.
Run dbt build
dbt build is the production command that runs all models (staging views, reporting marts) and all tests in the correct DAG order. If any model fails to compile or any test assertion breaks, the command stops and reports exactly what went wrong.
Before you run this, think back to Step 2. Last time you ran dbt tests after inserting raw data, they failed. This time you used entity resolution and survivorship rules. Do you expect all tests to pass now?
- Switch back to your terminal and navigate to your dbt project directory.
- Run the full build by executing this command:
dbt buildYou should see each model compiled and each test run. The final summary shows all models succeeded and all tests passed.
What does dbt build do?
dbt build combines dbt run (rebuilds all models) and dbt test (validates all assertions) into a single DAG-ordered execution. Models are rebuilt first, then their downstream tests run immediately after.
This is the same command you used in Project 2's Secret Mission. The difference now is that your production dimensions contain Vilandra's merged data, so the staging views and reporting marts pull in the new rows automatically.
Seeing test failures?
If a uniqueness test fails, the survivorship merge may have created a duplicate natural key. Run SELECT customer_id, COUNT(*) FROM dbo.DimCustomer GROUP BY customer_id HAVING COUNT(*) > 1 to find duplicates.
If a referential integrity test fails, check that the subsidiary_key on your Vilandra fact rows points to a valid DimSubsidiary row.
Help me debug my dbt test failures after the MDM merge.
Query the reporting marts to confirm Vilandra data flows through
Passing tests prove structural integrity. Now you need to confirm the actual data landed in your reporting marts.
- Return to your SQL query editor (SSMS or VS Code) and run this query against dbt_dev.cross_border_revenue:
SELECT *
FROM dbt_dev.cross_border_revenue
WHERE subsidiary_country = 'Lithuania';You should see rows showing Lithuania's revenue totals from the Vilandra sales orders you inserted in Step 4.
- Now query dbt_dev.customer_segment_profitability to confirm the new customer segments appear:
SELECT *
FROM dbt_dev.customer_segment_profitability
ORDER BY total_revenue DESC;You should see the Standard and Premium segments (from Vilandra's new customers) contributing revenue to the group-level totals.
Why does Lithuania appear automatically?
Your dbt models from Project 2 join FactSalesOrder to DimSubsidiary and aggregate by subsidiary_country. Because you inserted Vilandra's fact rows with the correct subsidiary_key pointing to the UAB Vilandra dimension row (country_code = 'LT'), the reporting marts pick up Lithuania without any model changes.
This is the payoff of a well-designed star schema. New acquisitions flow through existing reporting infrastructure as long as dimensions are resolved correctly.
Query returns no rows for Lithuania?
Check that the DimSubsidiary row has country_name = 'Lithuania' (the mart joins on this column). Run SELECT * FROM dbo.DimSubsidiary WHERE subsidiary_name = 'UAB Vilandra' to verify.
Also confirm your fact rows exist by running SELECT COUNT(*) FROM dbo.FactSalesOrder WHERE subsidiary_key = (SELECT subsidiary_key FROM dbo.DimSubsidiary WHERE subsidiary_name = 'UAB Vilandra'). You should see 3.
Help me figure out why Lithuania is missing from my cross_border_revenue mart.
Verify Power BI picks up the new data
The SQL queries proved the data exists in your reporting marts. Now confirm that Power BI (your Project 3 dashboard) reflects the new subsidiary visually.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Power BI Desktop and press Enter to open it.
- Open your BalticDW report file from Project 3.
- Click Refresh on the Home ribbon to re-import data from BalticDW.
After the refresh completes, check your country bar chart. Lithuania should now appear with revenue from the Vilandra sales orders. Navigate to your drillthrough page and confirm that the new Vilandra customers (Vilandra Beverages OOO, Panevezys Distribution, Kauno Tiekimas UAB) appear.
What does this prove?
You just validated the entire data platform end-to-end: acquisition CSV to entity resolution to survivorship merge to dbt transformation to Power BI visualization. A new subsidiary's data flows through every layer without a single schema change or model rewrite.
Lithuania not showing in your chart?
Make sure you clicked Refresh (not just reopened the file). Power BI caches data locally and only pulls new rows on an explicit refresh.
If the refresh fails with a connection error, confirm SQL Server is still running. Check Services (Windows) and verify that SQL Server (MSSQLSERVER) shows a Running status.
Help me troubleshoot why Power BI is not showing my new Vilandra data after refresh.
Secret mission
Secret Mission: Interactive Data Steward Review
The MDM pipeline quarantined records with scores between 70-89 because they are too ambiguous for auto-merge but too similar to ignore. Can you build a terminal-based review workflow that lets a data steward approve or reject each one?
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 on your local machine (SQL Server and Python), so there are no ongoing costs.
Resources you used:
- mdm schema with staging_customers, staging_products, quarantine, and merge_log tables (local SQL Server, no cost).
- UAB Vilandra data merged into dbo.DimCustomer, dbo.DimProduct, dbo.DimSubsidiary, and dbo.FactSalesOrder (local SQL Server, no cost).
- RapidFuzz Python package (installed via pip, no cost).
- Python scripts and CSV files: generate_vilandra.py, naive_load.py, entity_resolution.py, survivorship_merge.py, steward_review.py, setup_mdm_schema.sql, vilandra_customers.csv, vilandra_products.csv, resolution_results.json (local files, no cost).
- Updated dbt reporting marts in dbt_dev schema (local SQL Server, no cost).
Keep everything running
No action needed. Choose this if you plan to continue building on your BalticDW platform or want to experiment with the MDM pipeline further.
- Your BalticDW now contains 4 subsidiaries (Estonia, Latvia, Lithuania original + UAB Vilandra) and the MDM infrastructure is ready for future acquisitions.
- The RapidFuzz package remains installed in your Python environment.
- All Python scripts and CSV files remain in your project directory for reference.
Pause - I'll come back to this later
Shut down SQL Server to free up memory. All data persists on disk and resumes exactly where you left off when you restart the service.
- Open the Windows Services app (press the Windows key and type Services).
- Find SQL Server (MSSQLSERVER) in the list.
- Right-click it and select Stop.
- To resume later, right-click the same service and select Start.
Delete - I don't want to use this again
Remove all MDM project resources. You have two options depending on whether you want to keep Vilandra's merged data in your production dimensions.
Option A: Delete the MDM layer only (keep Vilandra data in production)
- Run this SQL in SSMS to drop the MDM tables and schema:
USE BalticDW;
GO
DROP TABLE IF EXISTS mdm.staging_customers;
DROP TABLE IF EXISTS mdm.staging_products;
DROP TABLE IF EXISTS mdm.quarantine;
DROP TABLE IF EXISTS mdm.merge_log;
DROP SCHEMA IF EXISTS mdm;
GOOption B: Revert Vilandra entirely (restore BalticDW to Project 3 state)
- Run this SQL in SSMS to remove Vilandra's data from your production dimensions and facts:
USE BalticDW;
GO
-- Remove Vilandra fact rows first (referential integrity)
DELETE FROM dbo.FactSalesOrder
WHERE subsidiary_key = (
SELECT subsidiary_key FROM dbo.DimSubsidiary
WHERE subsidiary_name = 'UAB Vilandra'
);
-- Remove Vilandra subsidiary
DELETE FROM dbo.DimSubsidiary WHERE subsidiary_name = 'UAB Vilandra';
-- Remove Vilandra customer and product rows
DELETE FROM dbo.DimCustomer WHERE customer_id LIKE 'VIL-%';
DELETE FROM dbo.DimProduct WHERE product_id LIKE 'VIL-%';
GO- Then drop the MDM schema using the Option A SQL above.
- Run dbt build from your dbt project directory to rebuild the reporting marts without Vilandra data.
Remove local files
- Delete the Python scripts, SQL file, and generated data files by running:
del generate_vilandra.py naive_load.py entity_resolution.py survivorship_merge.py steward_review.py setup_mdm_schema.sql vilandra_customers.csv vilandra_products.csv resolution_results.jsonUninstall RapidFuzz
- Remove the RapidFuzz package by running:
pip uninstall rapidfuzz- Type y when prompted to confirm the uninstall.
Nice Work!
Nice Work!
Nice work! You just built a production-style Master Data Management pipeline that onboards a new acquisition's data into an existing data warehouse using entity resolution, survivorship rules, and data quality gates.
You've learned how to:
- Built a fuzzy matching engine using RapidFuzz that resolves entity overlaps between acquisition data and canonical dimensions, using configurable similarity thresholds and a three-way classification (auto-match, quarantine, reject).
- Implemented per-attribute survivorship rules (source priority, completeness, recency) that create auditable golden records with a full merge log tracking every decision.
- Created a data quality gate with quarantine tables that prevent ambiguous or invalid records from reaching the production warehouse, then proved end-to-end platform integrity by running dbt build and seeing all tests pass with Vilandra's data flowing through to reporting marts.
- Secret Mission: Built an interactive data steward review workflow for quarantined records with approve/reject/skip controls and audit logging.
Ready to quiz yourself?