Enterprise Data Quality Validation Engine
Validate data integrity across SAP, Salesforce, and MDM with Python.
Introduction
30 Second Summary
When a company runs customer data through multiple business systems, a single typo or missing link between records can cascade into shipping errors, duplicate invoices, and lost revenue. Data quality engineers build the automated checks that catch these problems before they reach production.
In this project, you will build a validation engine that tests data flowing between three enterprise systems, catching duplicates, format mismatches, and broken cross-references. You will present your findings through a professional stakeholder dashboard.
What You'll Build
A Python-based enterprise data quality platform that generates synthetic customer records across three systems, validates cross-system data integrity with Great Expectations, runs end-to-end integration tests with root cause analysis, and displays quality metrics through a live Streamlit dashboard.
By the end of this project, you'll have:
- A synthetic data generator producing 1000+ customer records across SAP, Salesforce, and an MDM system with intentional anomalies injected at a controlled error rate.
- A validation suite and E2E integration test framework checking data quality across six dimensions including completeness, consistency, accuracy, and referential integrity.
- A root cause analysis module that classifies anomalies by type and severity, paired with a stakeholder-ready dashboard displaying quality scores and anomaly drill-downs.
- Secret Mission: An automated remediation engine that proposes and applies data cleanup actions with before/after comparison reports and audit logging.
Are there any prerequisites?
You should be comfortable with intermediate Python (classes, dictionaries, and pandas DataFrames). Familiarity with SQL concepts and a general understanding of what ERP/CRM systems do is helpful. No hands-on SAP or Salesforce access is needed.
Before We Start
Before We Start...
Set Up Your Python Environment and Tools
Before you can build a data quality engine, you need a clean Python environment with the right libraries installed.
In this step, you'll install Python 3.12, create an isolated virtual environment, and pull in the five key packages that power this project.
In this step, get ready to:
- Install Python 3.12 and create a virtual environment.
- Install project dependencies with pinned versions.
- Set up VS Code and verify all packages import correctly.
Install Python 3.12 and create a virtual environment
A virtual environment keeps your project's packages separate from your system Python. This prevents version conflicts between projects.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
- Check which version of Python is installed by running this command:
macOS
python3 --versionWindows
python --version✔️ I see version 3.12 or higher
Python 3.12 is ready to go. Skip ahead to the next instruction below.
ⓧ I see an older version
You have Python installed, but you need version 3.12 or higher for this project.
- Go to https://www.python.org/downloads and download the latest Python 3.12 installer for your operating system.
- Run the installer. On Windows, check the box that says Add python.exe to PATH before clicking Install Now.
- Close and reopen your terminal, then run the version check command again.
macOS users with an older system Python
macOS often ships with an older Python version. After installing Python 3.12, you may need to use python3.12 --version to confirm the new version. If your default python3 still points to the old version, use python3.12 for all commands in this project.
ⓧ Command not found
Python is not installed yet. Let's install it now.
- Go to https://www.python.org/downloads and download the latest Python 3.12 installer for your operating system.
- Run the installer. On Windows, check the box that says Add python.exe to PATH before clicking Install Now.
- On macOS, follow the prompts and accept the defaults.
- Close and reopen your terminal, then run the version check command again to confirm you see 3.12.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir data-quality-engine
cd data-quality-engine- Confirm the folder exists by running:
lsYou should see an empty directory since you just created it.
- Create a virtual environment by running this command:
macOS
python3 -m venv .venvWindows
python -m venv .venv- Activate the virtual environment by running this command:
macOS
source .venv/bin/activateWindows
.venv\Scripts\activateYou should see (.venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
What is a virtual environment?
A virtual environment is an isolated copy of Python just for this project. Packages you install inside it stay separate from your system Python and other projects.
The .venv folder holds this isolated environment. You activate it each time you work on the project.
Don't see (.venv) in your prompt?
On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try the activate command again.
On macOS, make sure you're using source before the path. Running the activate script directly without source will not work.
Help me activate my Python virtual environment
Install project dependencies
This project uses five key libraries. Pinning exact versions ensures everyone gets the same behavior regardless of when they run the install.
- Create a file called requirements.txt in your data-quality-engine folder by running this command:
macOS
touch requirements.txtWindows
New-Item requirements.txt- Confirm the file was created by running:
lsYou should see requirements.txt in the output.
- Open requirements.txt in any text editor and paste the following pinned dependency list:
great_expectations==1.3.10
pandas==2.2.3
pydantic==2.10.6
streamlit==1.42.2
faker==33.3.1What are these libraries?
- great_expectations is a data validation framework that defines and runs quality checks against your datasets.
- pandas provides DataFrames for loading, filtering, and transforming tabular data.
- pydantic enforces data contracts with type-checked models and field validators.
- streamlit builds interactive web dashboards from pure Python scripts.
- faker generates realistic synthetic data like names, addresses, and phone numbers.
- Save requirements.txt.
- Install all dependencies by running this command:
pip install -r requirements.txtThis will download and install all five packages plus their sub-dependencies. It may take 1-2 minutes.
Seeing a permission error or install failure?
Make sure your virtual environment is active. You should see (.venv) at the start of your terminal prompt. If not, run the activate command from the previous substep again.
If you see a network timeout, check your internet connection and try running the command again.
Help me fix my pip install error
✔️ 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
great_expectations==1.3.10
pandas==2.2.3
pydantic==2.10.6
streamlit==1.42.2
faker==33.3.1Set up VS Code and verify installations
Opening the project in VS Code with the right Python interpreter ensures code completion and linting work against your virtual environment's packages.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Visual Studio Code and press Enter to open it.
- Click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop and select the data-quality-engine folder, then click Open.
- Open the command palette by pressing Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows).
- Type Python: Select Interpreter and select it from the dropdown.
- Choose the interpreter that shows .venv in its path. It will look something like Python 3.12.x ('.venv').
Why select the .venv interpreter?
VS Code needs to know which Python to use for code analysis and running scripts. Selecting the virtual environment interpreter means VS Code sees all the packages you just installed. Without this, you would get false import errors in the editor.
- Open a terminal inside VS Code by pressing Ctrl+` (backtick key, same shortcut on macOS and Windows).
- Confirm your virtual environment is active. You should see (.venv) in the terminal prompt. If not, run the activate command from earlier.
- Verify all packages import correctly by running this command:
python -c "import great_expectations as gx; import pandas as pd; import pydantic; import streamlit; import faker; print('great_expectations:', gx.__version__); print('pandas:', pd.__version__); print('pydantic:', pydantic.__version__); print('streamlit:', streamlit.__version__); print('faker:', faker.__version__); print('All packages installed successfully!')"You should see version numbers printed for each package, followed by the message All packages installed successfully!.
Seeing a ModuleNotFoundError?
This means the terminal is using a different Python than the one with your installed packages. Make sure (.venv) appears in your prompt. If it does not, run the activate command for your platform.
If activation looks correct but packages are still missing, try running pip install -r requirements.txt again inside the VS Code terminal.
Help me fix my import errors
Your environment is locked and loaded. Next up, you'll design the data models that represent customer records across SAP, Salesforce, and MDM systems.
Design Synthetic Enterprise Data
Your environment is ready and all dependencies are installed. Now you need data to validate.
Enterprise systems like SAP, Salesforce, and MDM platforms each store customer records in their own format. When these systems exchange data, problems creep in: duplicate records, format mismatches, broken cross-references. In this step, you will generate realistic synthetic data that mimics these exact failure patterns so your validation engine has something meaningful to test against.
In this step, get ready to:
- Define Pydantic data models representing customer records in SAP, Salesforce, and MDM systems.
- Build a synthetic data generator that creates 1000+ cross-referenced records with deliberately injected anomalies.
- Run the generator to produce CSV files simulating real data extracts from each system.
Define the data architecture
Each enterprise system stores customer records differently. SAP uses a Business Partner format, Salesforce uses Account objects, and the MDM system maintains a "golden record" that reconciles the other two. You will define strict Pydantic schemas for each system so your validation engine knows exactly what valid data looks like.
- Create a new file called models.py in your project folder.
- Add the imports and the first data model by pasting this code:
"""Pydantic data models representing records in SAP, Salesforce, and MDM systems."""
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from datetime import date, datetime
class SAPBusinessPartner(BaseModel):
"""Customer record as stored in SAP ERP."""
bp_number: str = Field(description="SAP Business Partner number, format: BP-XXXXXXXX")
company_name: str
street: str
city: str
postal_code: str
country_code: str = Field(max_length=2)
phone: Optional[str] = None
email: Optional[str] = None
created_date: date
last_modified: datetime
is_active: bool = True
@field_validator("bp_number")
@classmethod
def validate_bp_number(cls, v: str) -> str:
if not v.startswith("BP-") or len(v) != 11:
raise ValueError("BP number must be format BP-XXXXXXXX (11 chars)")
return v
@field_validator("country_code")
@classmethod
def validate_country_code(cls, v: str) -> str:
if len(v) != 2 or not v.isalpha() or not v.isupper():
raise ValueError("Country code must be 2 uppercase letters")
return vWhat does this code do?
This defines the SAP Business Partner model as a Pydantic class with strict field types and format validators.
- bp_number must follow the pattern BP-XXXXXXXX (the field_validator enforces this at runtime).
- country_code is locked to exactly 2 uppercase letters, matching the ISO standard SAP uses.
- Optional fields like phone and email can be None, reflecting real-world data where these are often missing.
You won't see output yet. The next chunk adds the Salesforce model.
- Below the SAPBusinessPartner class, add the Salesforce Account model:
class SalesforceAccount(BaseModel):
"""Customer record as stored in Salesforce CRM."""
account_id: str = Field(description="Salesforce Account ID, format: SF-XXXXXXXXXX")
external_id: str = Field(description="Maps to SAP BP Number")
account_name: str
billing_street: str
billing_city: str
billing_postal_code: str
billing_country: str
phone: Optional[str] = None
email: Optional[str] = None
owner_id: str
created_date: datetime
last_activity_date: Optional[date] = None
status: str = "Active"
@field_validator("account_id")
@classmethod
def validate_account_id(cls, v: str) -> str:
if not v.startswith("SF-") or len(v) != 13:
raise ValueError("Account ID must be format SF-XXXXXXXXXX (13 chars)")
return vWhat does this code do?
The SalesforceAccount model mirrors how Salesforce stores customer data.
- external_id is the critical cross-reference field. It maps directly to SAP's bp_number. This is where mapping errors will show up during validation.
- Salesforce uses billing_ prefixes on address fields, while SAP uses plain names. This naming difference is deliberate and realistic.
You won't see output yet. The next chunk adds the MDM golden record model.
- Below the SalesforceAccount class, add the MDM Golden Record model:
class MDMGoldenRecord(BaseModel):
"""Master Data Management golden record - the single source of truth."""
master_id: str = Field(description="MDM Master ID, format: MDM-XXXXXXXXXXXX")
sap_bp_number: str
salesforce_account_id: str
canonical_name: str
canonical_street: str
canonical_city: str
canonical_postal_code: str
canonical_country: str
canonical_phone: Optional[str] = None
canonical_email: Optional[str] = None
confidence_score: float = Field(ge=0.0, le=1.0)
last_reconciled: datetime
source_priority: str = "SAP"
@field_validator("master_id")
@classmethod
def validate_master_id(cls, v: str) -> str:
if not v.startswith("MDM-") or len(v) != 16:
raise ValueError("Master ID must be format MDM-XXXXXXXXXXXX (16 chars)")
return vWhat does this code do?
The MDMGoldenRecord is the "single source of truth" that reconciles data from both SAP and Salesforce.
- It holds cross-references to both systems (sap_bp_number and salesforce_account_id). Broken links here indicate referential integrity failures.
- confidence_score (0.0 to 1.0) represents how certain the MDM system is that it matched the right records together.
- All address fields use the canonical_ prefix, indicating these are the "official" values resolved from potentially conflicting sources.
- Save models.py.
- Verify the models load correctly by running this in your terminal:
python models.pyYou should see no output and no errors. A clean exit means Python parsed all three models successfully.
Seeing a syntax error?
- Check that each class is separated by two blank lines at the module level (one blank line between the last method of one class and the class keyword of the next).
- Make sure the @field_validator decorators are indented inside the class body, not at module level.
- Confirm your virtual environment is active. If pydantic is not found, reactivate with .venv\Scripts\activate.
Help me fix a Pydantic model syntax error in models.py
✔️ Awesome, I've got everything!
Great. Double check you have saved models.py before continuing.
ⓧ I'd like to double check the full code
"""Pydantic data models representing records in SAP, Salesforce, and MDM systems."""
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from datetime import date, datetime
class SAPBusinessPartner(BaseModel):
"""Customer record as stored in SAP ERP."""
bp_number: str = Field(description="SAP Business Partner number, format: BP-XXXXXXXX")
company_name: str
street: str
city: str
postal_code: str
country_code: str = Field(max_length=2)
phone: Optional[str] = None
email: Optional[str] = None
created_date: date
last_modified: datetime
is_active: bool = True
@field_validator("bp_number")
@classmethod
def validate_bp_number(cls, v: str) -> str:
if not v.startswith("BP-") or len(v) != 11:
raise ValueError("BP number must be format BP-XXXXXXXX (11 chars)")
return v
@field_validator("country_code")
@classmethod
def validate_country_code(cls, v: str) -> str:
if len(v) != 2 or not v.isalpha() or not v.isupper():
raise ValueError("Country code must be 2 uppercase letters")
return v
class SalesforceAccount(BaseModel):
"""Customer record as stored in Salesforce CRM."""
account_id: str = Field(description="Salesforce Account ID, format: SF-XXXXXXXXXX")
external_id: str = Field(description="Maps to SAP BP Number")
account_name: str
billing_street: str
billing_city: str
billing_postal_code: str
billing_country: str
phone: Optional[str] = None
email: Optional[str] = None
owner_id: str
created_date: datetime
last_activity_date: Optional[date] = None
status: str = "Active"
@field_validator("account_id")
@classmethod
def validate_account_id(cls, v: str) -> str:
if not v.startswith("SF-") or len(v) != 13:
raise ValueError("Account ID must be format SF-XXXXXXXXXX (13 chars)")
return v
class MDMGoldenRecord(BaseModel):
"""Master Data Management golden record - the single source of truth."""
master_id: str = Field(description="MDM Master ID, format: MDM-XXXXXXXXXXXX")
sap_bp_number: str
salesforce_account_id: str
canonical_name: str
canonical_street: str
canonical_city: str
canonical_postal_code: str
canonical_country: str
canonical_phone: Optional[str] = None
canonical_email: Optional[str] = None
confidence_score: float = Field(ge=0.0, le=1.0)
last_reconciled: datetime
source_priority: str = "SAP"
@field_validator("master_id")
@classmethod
def validate_master_id(cls, v: str) -> str:
if not v.startswith("MDM-") or len(v) != 16:
raise ValueError("Master ID must be format MDM-XXXXXXXXXXXX (16 chars)")
return vBuild the synthetic data generator
Now you need realistic test data that exercises those models. The generator will create 1000 cross-referenced records across all three systems, then deliberately corrupt about 8% of them with anomalies like duplicates, format mismatches, and broken mappings.
- Create a new file called data_generator.py in your project folder.
- Add the imports, configuration, ID generators, and the anomaly injection function by pasting this code:
"""Synthetic data generator simulating SAP, Salesforce, and MDM records with anomalies."""
import random
import string
import csv
from datetime import date, datetime, timedelta
from pathlib import Path
from faker import Faker
fake = Faker()
Faker.seed(42)
random.seed(42)
OUTPUT_DIR = Path("data")
NUM_RECORDS = 1000
ANOMALY_RATE = 0.08
def generate_bp_number() -> str:
digits = "".join(random.choices(string.digits, k=8))
return f"BP-{digits}"
def generate_sf_id() -> str:
chars = "".join(random.choices(string.digits + string.ascii_uppercase, k=10))
return f"SF-{chars}"
def generate_mdm_id() -> str:
chars = "".join(random.choices(string.digits + string.ascii_uppercase, k=12))
return f"MDM-{chars}"
def inject_anomaly(record: dict, system: str) -> tuple[dict, str | None]:
"""Randomly inject an anomaly into a record. Returns (modified_record, anomaly_type)."""
anomaly_type = random.choice([
"duplicate", "format_mismatch", "missing_mapping",
"stale_record", "missing_field", "inconsistent_value"
])
if anomaly_type == "format_mismatch":
if system == "sap":
record["postal_code"] = record["postal_code"].replace("-", "").replace(" ", "")[:3]
elif system == "salesforce":
record["billing_country"] = record["billing_country"].lower()
elif anomaly_type == "missing_field":
if system == "sap":
record["email"] = ""
record["phone"] = ""
elif system == "salesforce":
record["phone"] = ""
elif anomaly_type == "inconsistent_value":
if system == "salesforce":
record["account_name"] = record["account_name"] + " Inc."
elif anomaly_type == "stale_record":
record["last_modified"] = (datetime.now() - timedelta(days=random.randint(365, 730))).isoformat()
elif anomaly_type == "missing_mapping":
if system == "mdm":
record["sap_bp_number"] = ""
elif anomaly_type == "duplicate":
pass # Duplicate handled at generation level
return record, anomaly_typeWhat does this code do?
This sets up the generator's foundation.
- Faker.seed(42) and random.seed(42) make the output reproducible. Every run produces the same records so your validation results are consistent.
- ANOMALY_RATE = 0.08 means about 8% of records get corrupted. This is realistic for enterprise integrations.
- The three ID generators (generate_bp_number, generate_sf_id, generate_mdm_id) produce IDs matching each system's format.
- inject_anomaly randomly picks from 6 anomaly types and applies the corruption. Each type targets specific fields that would realistically break in that way.
You won't see output yet. The next chunk adds the main generation logic.
- Below the inject_anomaly function, add the main dataset generation function, the CSV writer, and the entry point:
def generate_datasets():
"""Generate synthetic datasets for SAP, Salesforce, and MDM systems."""
OUTPUT_DIR.mkdir(exist_ok=True)
sap_records = []
sf_records = []
mdm_records = []
anomaly_log = []
bp_numbers = [generate_bp_number() for _ in range(NUM_RECORDS)]
sf_ids = [generate_sf_id() for _ in range(NUM_RECORDS)]
mdm_ids = [generate_mdm_id() for _ in range(NUM_RECORDS)]
for i in range(NUM_RECORDS):
company = fake.company()
street = fake.street_address()
city = fake.city()
postal_code = fake.postcode()
country = fake.country_code()
phone = fake.phone_number()
email = fake.company_email()
created = fake.date_between(start_date="-3y", end_date="-6m")
modified = fake.date_time_between(start_date="-6m", end_date="now")
sap_record = {
"bp_number": bp_numbers[i],
"company_name": company,
"street": street,
"city": city,
"postal_code": postal_code,
"country_code": country,
"phone": phone,
"email": email,
"created_date": created.isoformat(),
"last_modified": modified.isoformat(),
"is_active": "True"
}
sf_record = {
"account_id": sf_ids[i],
"external_id": bp_numbers[i],
"account_name": company,
"billing_street": street,
"billing_city": city,
"billing_postal_code": postal_code,
"billing_country": country,
"phone": phone,
"email": email,
"owner_id": f"USR-{random.randint(100, 999)}",
"created_date": modified.isoformat(),
"last_activity_date": fake.date_between(start_date="-30d", end_date="today").isoformat(),
"status": "Active"
}
mdm_record = {
"master_id": mdm_ids[i],
"sap_bp_number": bp_numbers[i],
"salesforce_account_id": sf_ids[i],
"canonical_name": company,
"canonical_street": street,
"canonical_city": city,
"canonical_postal_code": postal_code,
"canonical_country": country,
"canonical_phone": phone,
"canonical_email": email,
"confidence_score": round(random.uniform(0.7, 1.0), 2),
"last_reconciled": modified.isoformat(),
"source_priority": "SAP"
}
# Inject anomalies
if random.random() < ANOMALY_RATE:
system = random.choice(["sap", "salesforce", "mdm"])
if system == "sap":
sap_record, anomaly = inject_anomaly(sap_record, "sap")
elif system == "salesforce":
sf_record, anomaly = inject_anomaly(sf_record, "salesforce")
else:
mdm_record, anomaly = inject_anomaly(mdm_record, "mdm")
anomaly_log.append({
"record_index": i,
"system": system,
"anomaly_type": anomaly,
"bp_number": bp_numbers[i]
})
sap_records.append(sap_record)
sf_records.append(sf_record)
mdm_records.append(mdm_record)
# Handle duplicate injection
num_duplicates = int(NUM_RECORDS * 0.02)
for _ in range(num_duplicates):
idx = random.randint(0, NUM_RECORDS - 1)
dup_sap = sap_records[idx].copy()
dup_sap["bp_number"] = generate_bp_number()
sap_records.append(dup_sap)
anomaly_log.append({
"record_index": len(sap_records) - 1,
"system": "sap",
"anomaly_type": "duplicate",
"bp_number": dup_sap["bp_number"]
})
# Write CSVs
_write_csv(OUTPUT_DIR / "sap_business_partners.csv", sap_records)
_write_csv(OUTPUT_DIR / "salesforce_accounts.csv", sf_records)
_write_csv(OUTPUT_DIR / "mdm_golden_records.csv", mdm_records)
_write_csv(OUTPUT_DIR / "anomaly_log.csv", anomaly_log)
print(f"Generated {len(sap_records)} SAP records")
print(f"Generated {len(sf_records)} Salesforce records")
print(f"Generated {len(mdm_records)} MDM records")
print(f"Injected {len(anomaly_log)} anomalies")
def _write_csv(path: Path, records: list[dict]):
if not records:
return
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=records[0].keys())
writer.writeheader()
writer.writerows(records)
if __name__ == "__main__":
generate_datasets()What does this code do?
generate_datasets is the core function that builds all three datasets in a single pass.
- For each of the 1000 records, it generates one shared set of base data (company name, address, phone) and maps it into all three system formats. This ensures cross-references are valid by default.
- The external_id in Salesforce points to the same bp_number in SAP. The MDM record holds references to both. These are the links your validation engine will check.
- After the main loop, 2% additional duplicate SAP records are injected with new BP numbers but identical company data. This simulates the most common enterprise data problem.
- _write_csv saves each dataset as a CSV file inside a data/ folder.
- Save data_generator.py. You will run this file in the next substep.
Seeing an indentation error?
- The entire generate_datasets function body must be indented one level (4 spaces). If you pasted it with mixed tabs and spaces, select all and re-indent.
- Make sure the _write_csv function is at module level (no indentation), not nested inside generate_datasets.
Help me fix an indentation or paste error in data_generator.py
✔️ Awesome, I've got everything!
Great. Double check you have saved data_generator.py before continuing.
ⓧ I'd like to double check the full code
"""Synthetic data generator simulating SAP, Salesforce, and MDM records with anomalies."""
import random
import string
import csv
from datetime import date, datetime, timedelta
from pathlib import Path
from faker import Faker
fake = Faker()
Faker.seed(42)
random.seed(42)
OUTPUT_DIR = Path("data")
NUM_RECORDS = 1000
ANOMALY_RATE = 0.08
def generate_bp_number() -> str:
digits = "".join(random.choices(string.digits, k=8))
return f"BP-{digits}"
def generate_sf_id() -> str:
chars = "".join(random.choices(string.digits + string.ascii_uppercase, k=10))
return f"SF-{chars}"
def generate_mdm_id() -> str:
chars = "".join(random.choices(string.digits + string.ascii_uppercase, k=12))
return f"MDM-{chars}"
def inject_anomaly(record: dict, system: str) -> tuple[dict, str | None]:
"""Randomly inject an anomaly into a record. Returns (modified_record, anomaly_type)."""
anomaly_type = random.choice([
"duplicate", "format_mismatch", "missing_mapping",
"stale_record", "missing_field", "inconsistent_value"
])
if anomaly_type == "format_mismatch":
if system == "sap":
record["postal_code"] = record["postal_code"].replace("-", "").replace(" ", "")[:3]
elif system == "salesforce":
record["billing_country"] = record["billing_country"].lower()
elif anomaly_type == "missing_field":
if system == "sap":
record["email"] = ""
record["phone"] = ""
elif system == "salesforce":
record["phone"] = ""
elif anomaly_type == "inconsistent_value":
if system == "salesforce":
record["account_name"] = record["account_name"] + " Inc."
elif anomaly_type == "stale_record":
record["last_modified"] = (datetime.now() - timedelta(days=random.randint(365, 730))).isoformat()
elif anomaly_type == "missing_mapping":
if system == "mdm":
record["sap_bp_number"] = ""
elif anomaly_type == "duplicate":
pass # Duplicate handled at generation level
return record, anomaly_type
def generate_datasets():
"""Generate synthetic datasets for SAP, Salesforce, and MDM systems."""
OUTPUT_DIR.mkdir(exist_ok=True)
sap_records = []
sf_records = []
mdm_records = []
anomaly_log = []
bp_numbers = [generate_bp_number() for _ in range(NUM_RECORDS)]
sf_ids = [generate_sf_id() for _ in range(NUM_RECORDS)]
mdm_ids = [generate_mdm_id() for _ in range(NUM_RECORDS)]
for i in range(NUM_RECORDS):
company = fake.company()
street = fake.street_address()
city = fake.city()
postal_code = fake.postcode()
country = fake.country_code()
phone = fake.phone_number()
email = fake.company_email()
created = fake.date_between(start_date="-3y", end_date="-6m")
modified = fake.date_time_between(start_date="-6m", end_date="now")
sap_record = {
"bp_number": bp_numbers[i],
"company_name": company,
"street": street,
"city": city,
"postal_code": postal_code,
"country_code": country,
"phone": phone,
"email": email,
"created_date": created.isoformat(),
"last_modified": modified.isoformat(),
"is_active": "True"
}
sf_record = {
"account_id": sf_ids[i],
"external_id": bp_numbers[i],
"account_name": company,
"billing_street": street,
"billing_city": city,
"billing_postal_code": postal_code,
"billing_country": country,
"phone": phone,
"email": email,
"owner_id": f"USR-{random.randint(100, 999)}",
"created_date": modified.isoformat(),
"last_activity_date": fake.date_between(start_date="-30d", end_date="today").isoformat(),
"status": "Active"
}
mdm_record = {
"master_id": mdm_ids[i],
"sap_bp_number": bp_numbers[i],
"salesforce_account_id": sf_ids[i],
"canonical_name": company,
"canonical_street": street,
"canonical_city": city,
"canonical_postal_code": postal_code,
"canonical_country": country,
"canonical_phone": phone,
"canonical_email": email,
"confidence_score": round(random.uniform(0.7, 1.0), 2),
"last_reconciled": modified.isoformat(),
"source_priority": "SAP"
}
# Inject anomalies
if random.random() < ANOMALY_RATE:
system = random.choice(["sap", "salesforce", "mdm"])
if system == "sap":
sap_record, anomaly = inject_anomaly(sap_record, "sap")
elif system == "salesforce":
sf_record, anomaly = inject_anomaly(sf_record, "salesforce")
else:
mdm_record, anomaly = inject_anomaly(mdm_record, "mdm")
anomaly_log.append({
"record_index": i,
"system": system,
"anomaly_type": anomaly,
"bp_number": bp_numbers[i]
})
sap_records.append(sap_record)
sf_records.append(sf_record)
mdm_records.append(mdm_record)
# Handle duplicate injection
num_duplicates = int(NUM_RECORDS * 0.02)
for _ in range(num_duplicates):
idx = random.randint(0, NUM_RECORDS - 1)
dup_sap = sap_records[idx].copy()
dup_sap["bp_number"] = generate_bp_number()
sap_records.append(dup_sap)
anomaly_log.append({
"record_index": len(sap_records) - 1,
"system": "sap",
"anomaly_type": "duplicate",
"bp_number": dup_sap["bp_number"]
})
# Write CSVs
_write_csv(OUTPUT_DIR / "sap_business_partners.csv", sap_records)
_write_csv(OUTPUT_DIR / "salesforce_accounts.csv", sf_records)
_write_csv(OUTPUT_DIR / "mdm_golden_records.csv", mdm_records)
_write_csv(OUTPUT_DIR / "anomaly_log.csv", anomaly_log)
print(f"Generated {len(sap_records)} SAP records")
print(f"Generated {len(sf_records)} Salesforce records")
print(f"Generated {len(mdm_records)} MDM records")
print(f"Injected {len(anomaly_log)} anomalies")
def _write_csv(path: Path, records: list[dict]):
if not records:
return
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=records[0].keys())
writer.writeheader()
writer.writerows(records)
if __name__ == "__main__":
generate_datasets()Generate and persist the test datasets
The generator is ready. Time to run it and produce the CSV files that simulate real system extracts.
- Run the data generator by executing this command:
python data_generator.py- Confirm the output files were created by running:
ls data/You should see four files listed: sap_business_partners.csv, salesforce_accounts.csv, mdm_golden_records.csv, and anomaly_log.csv.
Your terminal should also show output like this:
Generated 1020 SAP records
Generated 1000 Salesforce records
Generated 1000 MDM records
Injected 100 anomaliesWhy 1020 SAP records instead of 1000?
The base loop creates 1000 records for each system. After that, 2% additional duplicate SAP records are appended (about 20). These duplicates have new BP numbers but identical company data, simulating the most common data quality problem in enterprise systems.
The exact anomaly count may vary slightly based on the random seed hitting the 8% threshold. You should see a number close to 100 total (80 from the main loop plus 20 duplicates).
Seeing a ModuleNotFoundError for faker?
- Confirm your virtual environment is active. Your terminal prompt should show (.venv) at the beginning.
- If the environment is not active, run .venv\Scripts\activate and try again.
- If the error persists, reinstall the dependency with pip install faker.
Help me fix a ModuleNotFoundError when running data_generator.py
Your synthetic enterprise data is ready. You now have realistic test datasets with deliberate anomalies waiting to be caught. Next up, you will build the validation engine that detects these data quality issues using Great Expectations.
Build the Cross-System Validation Engine
Your synthetic datasets are ready with realistic enterprise records and intentional anomalies scattered throughout. Now you need a way to systematically test data quality across all three systems.
In this step, you will build a validation engine using Great Expectations that checks individual system data for completeness, format, and uniqueness. It also verifies cross-system consistency by confirming that IDs, names, and addresses match between SAP, Salesforce, and MDM.
In this step, get ready to:
- Initialize a Great Expectations Data Context that persists validation metadata to disk.
- Define expectation suites covering six data quality dimensions across all three systems.
- Run all validation suites and generate structured JSON results for downstream reporting.
Initialize a Great Expectations Data Context
A Data Context is the central configuration hub for Great Expectations. It stores metadata about your data sources, expectation suites, and validation results as YAML files on disk.
- Create a new file called validation_engine.py in your data-quality-engine project folder.
- Add the imports, path constants, and context setup function by pasting this code:
"""Great Expectations validation engine for cross-system data quality checks."""
import great_expectations as gx
from great_expectations.core import ExpectationSuiteConfig
import pandas as pd
from pathlib import Path
import json
DATA_DIR = Path("data")
RESULTS_DIR = Path("results")
def setup_context() -> gx.DataContext:
"""Initialize a GX File Data Context."""
context = gx.get_context(mode="file")
return contextWhat does this code do?
The gx.get_context(mode="file") call creates (or reconnects to) a File Data Context in your project. This creates a gx/ directory that persists configuration between runs.
DATA_DIR points to where your generated CSVs live. RESULTS_DIR is where validation output will be saved later.
- Save validation_engine.py.
Why Great Expectations over manual checks?
You could write raw pandas assertions, but Great Expectations gives you a declarative vocabulary of 150+ built-in checks. Each check produces structured metadata (pass/fail counts, observed values) that feeds directly into reports and dashboards.
GX also generates Data Docs automatically. These are static HTML pages showing every expectation's result in a stakeholder-friendly format.
Define cross-system expectation suites
Each validation function connects a pandas DataFrame to the Data Context, defines expectations (the checks), and runs them against a batch of data. You will create four suites: one per system plus a cross-system consistency suite.
- Add the SAP validation function below your setup_context function in validation_engine.py:
def validate_sap_data(context: gx.DataContext) -> dict:
"""Validate SAP Business Partner data quality."""
df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
data_source = context.data_sources.add_pandas("sap_source")
data_asset = data_source.add_dataframe_asset(name="sap_partners")
batch_definition = data_asset.add_batch_definition_whole_dataframe("sap_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
suite = gx.ExpectationSuite(name="sap_quality_suite")
# Completeness checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="bp_number")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="company_name")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="country_code")
)
# Format checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="bp_number", regex=r"^BP-\d{8}$"
)
)
suite.add_expectation(
gx.expectations.ExpectColumnValueLengthsToBeBetween(
column="country_code", min_value=2, max_value=2
)
)
# Uniqueness checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeUnique(column="bp_number")
)
# Validity checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeInSet(
column="is_active", value_set=["True", "False"]
)
)
results = batch.validate(suite)
return _extract_results(results, "SAP Business Partners")What does this code do?
This function covers four data quality dimensions for SAP records:
- Completeness checks ensure critical fields (bp_number, company_name, country_code) are never null.
- Format checks verify that bp_number matches the BP-XXXXXXXX pattern and country_code is exactly 2 characters.
- Uniqueness checks confirm no duplicate bp_number values exist.
- Validity checks confirm is_active contains only allowed values.
The context.data_sources.add_pandas() call registers a pandas data source. The add_batch_definition_whole_dataframe() method tells GX to treat the entire DataFrame as one batch for validation.
- Add the Salesforce validation function below validate_sap_data:
def validate_salesforce_data(context: gx.DataContext) -> dict:
"""Validate Salesforce Account data quality."""
df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
data_source = context.data_sources.add_pandas("sf_source")
data_asset = data_source.add_dataframe_asset(name="sf_accounts")
batch_definition = data_asset.add_batch_definition_whole_dataframe("sf_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
suite = gx.ExpectationSuite(name="salesforce_quality_suite")
# Completeness
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="account_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="external_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="account_name")
)
# Format
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="account_id", regex=r"^SF-[A-Z0-9]{10}$"
)
)
# Referential integrity (external_id should match SAP format)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="external_id", regex=r"^BP-\d{8}$"
)
)
# Status validity
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeInSet(
column="status", value_set=["Active", "Inactive", "Prospect"]
)
)
results = batch.validate(suite)
return _extract_results(results, "Salesforce Accounts")What does this code do?
The Salesforce suite validates completeness, format, and referential integrity. The external_id regex check is important because it verifies that every Salesforce account references a valid SAP BP number format. If the external_id doesn't match ^BP-\d{8}$, the integration link is broken.
- Add the MDM validation function below validate_salesforce_data:
def validate_mdm_data(context: gx.DataContext) -> dict:
"""Validate MDM Golden Record data quality."""
df = pd.read_csv(DATA_DIR / "mdm_golden_records.csv")
data_source = context.data_sources.add_pandas("mdm_source")
data_asset = data_source.add_dataframe_asset(name="mdm_records")
batch_definition = data_asset.add_batch_definition_whole_dataframe("mdm_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
suite = gx.ExpectationSuite(name="mdm_quality_suite")
# Completeness
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="master_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="sap_bp_number")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="salesforce_account_id")
)
# Format
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="master_id", regex=r"^MDM-[A-Z0-9]{12}$"
)
)
# Confidence score range
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeBetween(
column="confidence_score", min_value=0.0, max_value=1.0
)
)
results = batch.validate(suite)
return _extract_results(results, "MDM Golden Records")What does this code do?
The MDM suite checks that every golden record has complete cross-references (sap_bp_number and salesforce_account_id both present). It also validates that confidence_score stays within the 0.0 to 1.0 range. Records that fail this check indicate corruption in the reconciliation process.
- Add the cross-system consistency function below validate_mdm_data:
def validate_cross_system_consistency() -> dict:
"""Validate referential integrity across all three systems."""
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sf_df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
mdm_df = pd.read_csv(DATA_DIR / "mdm_golden_records.csv")
results = {
"system": "Cross-System",
"checks": [],
"passed": 0,
"failed": 0,
"total": 0
}
# Check: All SF external_ids exist in SAP
sap_bps = set(sap_df["bp_number"].dropna())
sf_externals = set(sf_df["external_id"].dropna())
orphaned_sf = sf_externals - sap_bps
check_passed = len(orphaned_sf) == 0
results["checks"].append({
"name": "SF external_ids reference valid SAP BP numbers",
"passed": check_passed,
"details": f"{len(orphaned_sf)} orphaned references found" if not check_passed else "All references valid"
})
# Check: MDM records reference valid SAP and SF IDs
mdm_sap_refs = set(mdm_df["sap_bp_number"].dropna()) - {""}
orphaned_mdm_sap = mdm_sap_refs - sap_bps
check_passed = len(orphaned_mdm_sap) == 0
results["checks"].append({
"name": "MDM sap_bp_numbers reference valid SAP records",
"passed": check_passed,
"details": f"{len(orphaned_mdm_sap)} orphaned MDM->SAP references" if not check_passed else "All references valid"
})
# Check: Name consistency between systems
merged = sap_df.merge(sf_df, left_on="bp_number", right_on="external_id", how="inner")
name_mismatches = merged[merged["company_name"] != merged["account_name"]]
check_passed = len(name_mismatches) == 0
results["checks"].append({
"name": "Company names consistent between SAP and Salesforce",
"passed": check_passed,
"details": f"{len(name_mismatches)} name mismatches detected" if not check_passed else "All names consistent"
})
# Check: Address consistency
address_mismatches = merged[merged["city"] != merged["billing_city"]]
check_passed = len(address_mismatches) == 0
results["checks"].append({
"name": "City values consistent between SAP and Salesforce",
"passed": check_passed,
"details": f"{len(address_mismatches)} city mismatches" if not check_passed else "All cities consistent"
})
for check in results["checks"]:
results["total"] += 1
if check["passed"]:
results["passed"] += 1
else:
results["failed"] += 1
return resultsWhat does this code do?
This function performs checks that span multiple systems. It uses set operations and DataFrame merges instead of GX expectations because these checks compare data across files rather than validating one file in isolation.
- Referential integrity confirms that every Salesforce external_id and MDM sap_bp_number points to a real SAP record.
- Consistency checks compare company_name and city values across matched records. Mismatches reveal data drift between the systems.
- Save validation_engine.py.
Run validation checkpoints and generate results
Now you need two more functions: a helper that extracts structured results from GX's validation output, and an orchestrator that runs everything and writes the JSON report.
- Add the result extraction helper below validate_cross_system_consistency in validation_engine.py:
def _extract_results(validation_result, system_name: str) -> dict:
"""Extract structured results from a GX validation result."""
results = {
"system": system_name,
"success": validation_result.success,
"checks": [],
"passed": 0,
"failed": 0,
"total": 0
}
for result in validation_result.results:
check = {
"expectation_type": result.expectation_config.type,
"column": result.expectation_config.kwargs.get("column", "N/A"),
"success": result.success,
"observed_value": str(result.result.get("observed_value", "N/A"))
}
results["checks"].append(check)
results["total"] += 1
if result.success:
results["passed"] += 1
else:
results["failed"] += 1
return resultsWhat does this code do?
This helper iterates through GX's native validation result object and extracts the expectation type, target column, pass/fail status, and observed value for each check. The output is a plain dictionary that can be serialized to JSON for your dashboard to consume.
- Add the orchestrator function and the entry point below _extract_results:
def run_all_validations() -> list[dict]:
"""Run all validation suites and return results."""
RESULTS_DIR.mkdir(exist_ok=True)
context = setup_context()
all_results = []
all_results.append(validate_sap_data(context))
all_results.append(validate_salesforce_data(context))
all_results.append(validate_mdm_data(context))
all_results.append(validate_cross_system_consistency())
with open(RESULTS_DIR / "validation_results.json", "w") as f:
json.dump(all_results, f, indent=2, default=str)
print("Validation complete. Results saved to results/validation_results.json")
for r in all_results:
status = "PASS" if r.get("success", r["failed"] == 0) else "FAIL"
print(f" [{status}] {r['system']}: {r['passed']}/{r['total']} checks passed")
return all_results
if __name__ == "__main__":
run_all_validations()What does this code do?
run_all_validations() creates the results/ directory, initializes the Data Context, and runs all four validation suites in sequence. It writes the combined results to results/validation_results.json and prints a summary to the terminal.
- Save validation_engine.py.
- Run the validation engine by executing this command in your terminal:
python validation_engine.py- You should see output listing each system's pass/fail status. Some checks will fail because of the anomalies you injected in the previous step.
What should I see?
Your terminal output will look something like:
Validation complete. Results saved to results/validation_results.json
[FAIL] SAP Business Partners: 5/7 checks passed
[FAIL] Salesforce Accounts: 5/6 checks passed
[FAIL] MDM Golden Records: 4/5 checks passed
[FAIL] Cross-System: 2/4 checks passedThe exact numbers depend on which anomalies were randomly assigned, but you will see failures. This confirms your validation engine is detecting the injected data quality issues.
Seeing an import error or missing module?
Make sure your virtual environment is active. On Windows, run .venv\Scripts\activate before running the script.
If you see a FileNotFoundError for the CSV files, confirm you ran data_generator.py in the previous step and that the data/ directory contains your CSV files.
Help me debug my validation engine
- Confirm the results/ directory was created by running:
ls results/- You should see validation_results.json listed. You will also see a gx/ directory in your project root where Great Expectations stores its configuration.
✔️ Awesome, I've got everything!
Great. Double check you have saved validation_engine.py and that results/validation_results.json exists.
ⓧ I'd like to double check the full code
Here is the complete validation_engine.py file:
"""Great Expectations validation engine for cross-system data quality checks."""
import great_expectations as gx
from great_expectations.core import ExpectationSuiteConfig
import pandas as pd
from pathlib import Path
import json
DATA_DIR = Path("data")
RESULTS_DIR = Path("results")
def setup_context() -> gx.DataContext:
"""Initialize a GX File Data Context."""
context = gx.get_context(mode="file")
return context
def validate_sap_data(context: gx.DataContext) -> dict:
"""Validate SAP Business Partner data quality."""
df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
data_source = context.data_sources.add_pandas("sap_source")
data_asset = data_source.add_dataframe_asset(name="sap_partners")
batch_definition = data_asset.add_batch_definition_whole_dataframe("sap_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
suite = gx.ExpectationSuite(name="sap_quality_suite")
# Completeness checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="bp_number")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="company_name")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="country_code")
)
# Format checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="bp_number", regex=r"^BP-\d{8}$"
)
)
suite.add_expectation(
gx.expectations.ExpectColumnValueLengthsToBeBetween(
column="country_code", min_value=2, max_value=2
)
)
# Uniqueness checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeUnique(column="bp_number")
)
# Validity checks
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeInSet(
column="is_active", value_set=["True", "False"]
)
)
results = batch.validate(suite)
return _extract_results(results, "SAP Business Partners")
def validate_salesforce_data(context: gx.DataContext) -> dict:
"""Validate Salesforce Account data quality."""
df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
data_source = context.data_sources.add_pandas("sf_source")
data_asset = data_source.add_dataframe_asset(name="sf_accounts")
batch_definition = data_asset.add_batch_definition_whole_dataframe("sf_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
suite = gx.ExpectationSuite(name="salesforce_quality_suite")
# Completeness
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="account_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="external_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="account_name")
)
# Format
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="account_id", regex=r"^SF-[A-Z0-9]{10}$"
)
)
# Referential integrity (external_id should match SAP format)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="external_id", regex=r"^BP-\d{8}$"
)
)
# Status validity
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeInSet(
column="status", value_set=["Active", "Inactive", "Prospect"]
)
)
results = batch.validate(suite)
return _extract_results(results, "Salesforce Accounts")
def validate_mdm_data(context: gx.DataContext) -> dict:
"""Validate MDM Golden Record data quality."""
df = pd.read_csv(DATA_DIR / "mdm_golden_records.csv")
data_source = context.data_sources.add_pandas("mdm_source")
data_asset = data_source.add_dataframe_asset(name="mdm_records")
batch_definition = data_asset.add_batch_definition_whole_dataframe("mdm_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
suite = gx.ExpectationSuite(name="mdm_quality_suite")
# Completeness
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="master_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="sap_bp_number")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="salesforce_account_id")
)
# Format
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="master_id", regex=r"^MDM-[A-Z0-9]{12}$"
)
)
# Confidence score range
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeBetween(
column="confidence_score", min_value=0.0, max_value=1.0
)
)
results = batch.validate(suite)
return _extract_results(results, "MDM Golden Records")
def validate_cross_system_consistency() -> dict:
"""Validate referential integrity across all three systems."""
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sf_df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
mdm_df = pd.read_csv(DATA_DIR / "mdm_golden_records.csv")
results = {
"system": "Cross-System",
"checks": [],
"passed": 0,
"failed": 0,
"total": 0
}
# Check: All SF external_ids exist in SAP
sap_bps = set(sap_df["bp_number"].dropna())
sf_externals = set(sf_df["external_id"].dropna())
orphaned_sf = sf_externals - sap_bps
check_passed = len(orphaned_sf) == 0
results["checks"].append({
"name": "SF external_ids reference valid SAP BP numbers",
"passed": check_passed,
"details": f"{len(orphaned_sf)} orphaned references found" if not check_passed else "All references valid"
})
# Check: MDM records reference valid SAP and SF IDs
mdm_sap_refs = set(mdm_df["sap_bp_number"].dropna()) - {""}
orphaned_mdm_sap = mdm_sap_refs - sap_bps
check_passed = len(orphaned_mdm_sap) == 0
results["checks"].append({
"name": "MDM sap_bp_numbers reference valid SAP records",
"passed": check_passed,
"details": f"{len(orphaned_mdm_sap)} orphaned MDM->SAP references" if not check_passed else "All references valid"
})
# Check: Name consistency between systems
merged = sap_df.merge(sf_df, left_on="bp_number", right_on="external_id", how="inner")
name_mismatches = merged[merged["company_name"] != merged["account_name"]]
check_passed = len(name_mismatches) == 0
results["checks"].append({
"name": "Company names consistent between SAP and Salesforce",
"passed": check_passed,
"details": f"{len(name_mismatches)} name mismatches detected" if not check_passed else "All names consistent"
})
# Check: Address consistency
address_mismatches = merged[merged["city"] != merged["billing_city"]]
check_passed = len(address_mismatches) == 0
results["checks"].append({
"name": "City values consistent between SAP and Salesforce",
"passed": check_passed,
"details": f"{len(address_mismatches)} city mismatches" if not check_passed else "All cities consistent"
})
for check in results["checks"]:
results["total"] += 1
if check["passed"]:
results["passed"] += 1
else:
results["failed"] += 1
return results
def _extract_results(validation_result, system_name: str) -> dict:
"""Extract structured results from a GX validation result."""
results = {
"system": system_name,
"success": validation_result.success,
"checks": [],
"passed": 0,
"failed": 0,
"total": 0
}
for result in validation_result.results:
check = {
"expectation_type": result.expectation_config.type,
"column": result.expectation_config.kwargs.get("column", "N/A"),
"success": result.success,
"observed_value": str(result.result.get("observed_value", "N/A"))
}
results["checks"].append(check)
results["total"] += 1
if result.success:
results["passed"] += 1
else:
results["failed"] += 1
return results
def run_all_validations() -> list[dict]:
"""Run all validation suites and return results."""
RESULTS_DIR.mkdir(exist_ok=True)
context = setup_context()
all_results = []
all_results.append(validate_sap_data(context))
all_results.append(validate_salesforce_data(context))
all_results.append(validate_mdm_data(context))
all_results.append(validate_cross_system_consistency())
with open(RESULTS_DIR / "validation_results.json", "w") as f:
json.dump(all_results, f, indent=2, default=str)
print("Validation complete. Results saved to results/validation_results.json")
for r in all_results:
status = "PASS" if r.get("success", r["failed"] == 0) else "FAIL"
print(f" [{status}] {r['system']}: {r['passed']}/{r['total']} checks passed")
return all_results
if __name__ == "__main__":
run_all_validations()Your validation engine is working and detecting real data quality issues across all three systems. Next, you will build end-to-end integration tests that simulate the full journey of a record from SAP through MDM to Salesforce.
Implement E2E Integration Tests
Your validation engine now checks data quality within each system and flags cross-system inconsistencies. But those checks are static. They examine data at rest, not data in motion.
In a real enterprise, a customer record travels from SAP through a transformation layer into Salesforce, with the MDM system reconciling both. Each hop can introduce errors. In this step, you will build E2E integration tests that simulate that journey and verify data integrity at every stage.
In this step, get ready to:
- Design test scenarios that track a record's journey across SAP, Salesforce, and MDM.
- Build transformation validators that compare source and target records after each hop.
- Execute the full test suite and generate a structured JSON report for stakeholder review.
Design test scenarios for data transformation flows
Each E2E test follows the same pattern: load data from two systems, compare records that should match, and report how many records failed the check. You need a container class that standardizes how every test stores its name, pass/fail status, and affected record count.
- Create a new file called e2e_tests.py in your project root (the same folder as validation_engine.py).
- Add the imports, constants, and the E2ETestResult class by pasting this code:
"""End-to-end integration test scenarios for cross-system data flows."""
import pandas as pd
from pathlib import Path
from datetime import datetime
import json
DATA_DIR = Path("data")
RESULTS_DIR = Path("results")
class E2ETestResult:
"""Container for a single E2E test result."""
def __init__(self, name: str, scenario: str):
self.name = name
self.scenario = scenario
self.passed = False
self.details = ""
self.affected_records = 0
self.timestamp = datetime.now().isoformat()
def to_dict(self) -> dict:
return {
"name": self.name,
"scenario": self.scenario,
"passed": self.passed,
"details": self.details,
"affected_records": self.affected_records,
"timestamp": self.timestamp
}What does this code do?
DATA_DIR and RESULTS_DIR point to the folders holding your generated CSVs and output reports.
E2ETestResult is a structured container for test outcomes. Every test function you write will create one of these, set its passed flag and details string, and return it. The to_dict() method converts the result to JSON-serializable format for the final report.
- Save e2e_tests.py.
You won't see a change yet. The next chunks use this class to build the actual test functions.
Build the transformation validator
Each test function simulates one validation scenario that an enterprise data team runs during integration testing. The first two tests focus on the SAP-to-Salesforce mapping: do all active SAP records have a Salesforce counterpart, and do the field values match after transformation?
- In e2e_tests.py, add the SAP-to-Salesforce mapping test and the field transformation accuracy test below the E2ETestResult class:
def test_sap_to_salesforce_mapping() -> E2ETestResult:
"""Test that SAP records correctly map to Salesforce via external_id."""
result = E2ETestResult(
name="SAP to Salesforce Mapping",
scenario="Verify all SAP BP numbers have corresponding Salesforce accounts"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sf_df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
# Filter to active SAP records only
sap_active = sap_df[sap_df["is_active"] == True]
# Left join to find SAP records with no Salesforce match
mapped = sap_active.merge(sf_df, left_on="bp_number", right_on="external_id", how="left")
unmapped = mapped[mapped["account_id"].isna()]
result.affected_records = len(unmapped)
result.passed = len(unmapped) == 0
result.details = (
f"{len(unmapped)} active SAP records have no Salesforce account mapping"
if not result.passed else "All active SAP records mapped to Salesforce"
)
return result
def test_field_transformation_accuracy() -> E2ETestResult:
"""Test that field values are correctly transformed between systems."""
result = E2ETestResult(
name="Field Transformation Accuracy",
scenario="Verify field values match after SAP->Salesforce transformation"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sf_df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
# Join on the mapping key
merged = sap_df.merge(sf_df, left_on="bp_number", right_on="external_id", how="inner")
# Define expected field-to-field mappings
field_mappings = [
("company_name", "account_name"),
("street", "billing_street"),
("city", "billing_city"),
("postal_code", "billing_postal_code"),
]
mismatches = 0
mismatch_details = []
for sap_field, sf_field in field_mappings:
diff = merged[merged[sap_field] != merged[sf_field]]
if len(diff) > 0:
mismatches += len(diff)
mismatch_details.append(f"{sap_field}->{sf_field}: {len(diff)} mismatches")
result.affected_records = mismatches
result.passed = mismatches == 0
result.details = (
"; ".join(mismatch_details) if mismatch_details
else "All field transformations accurate"
)
return resultWhat do these tests check?
test_sap_to_salesforce_mapping() filters SAP data to active records, then left-joins against Salesforce on the bp_number/external_id key. Any SAP record without a Salesforce counterpart is an orphan. This catches broken integration links.
test_field_transformation_accuracy() compares four field pairs between the joined SAP and Salesforce records. If a company's name, street, city, or postal code differ between the two systems, the transformation introduced an error.
- Save e2e_tests.py.
You won't see output yet. The next chunk adds the remaining three test functions.
The remaining tests cover the MDM system's cross-references, data freshness, and duplicate detection. These round out the test suite so you are checking all five data quality dimensions relevant to the integration flow.
- Add the MDM completeness, data freshness, and duplicate detection tests below test_field_transformation_accuracy():
def test_mdm_completeness() -> E2ETestResult:
"""Test that MDM golden records have complete cross-references."""
result = E2ETestResult(
name="MDM Cross-Reference Completeness",
scenario="Verify MDM records have both SAP and Salesforce references"
)
mdm_df = pd.read_csv(DATA_DIR / "mdm_golden_records.csv")
# Find records missing SAP or Salesforce links
missing_sap = mdm_df[
(mdm_df["sap_bp_number"].isna()) | (mdm_df["sap_bp_number"] == "")
]
missing_sf = mdm_df[
(mdm_df["salesforce_account_id"].isna()) | (mdm_df["salesforce_account_id"] == "")
]
total_incomplete = len(missing_sap) + len(missing_sf)
result.affected_records = total_incomplete
result.passed = total_incomplete == 0
result.details = (
f"Missing SAP refs: {len(missing_sap)}, Missing SF refs: {len(missing_sf)}"
if not result.passed else "All MDM records have complete cross-references"
)
return result
def test_data_freshness() -> E2ETestResult:
"""Test that records are not stale (modified within last 180 days)."""
result = E2ETestResult(
name="Data Freshness",
scenario="Verify records have been updated within the last 180 days"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sap_df["last_modified"] = pd.to_datetime(sap_df["last_modified"])
# Records older than 180 days are considered stale
threshold = datetime.now() - pd.Timedelta(days=180)
stale = sap_df[sap_df["last_modified"] < threshold]
result.affected_records = len(stale)
result.passed = len(stale) == 0
result.details = (
f"{len(stale)} records not updated in 180+ days"
if not result.passed else "All records are fresh"
)
return result
def test_duplicate_detection() -> E2ETestResult:
"""Test for duplicate records based on company name similarity."""
result = E2ETestResult(
name="Duplicate Detection",
scenario="Identify records with identical company names across different BP numbers"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
# Find rows where company_name appears more than once
duplicates = sap_df[sap_df.duplicated(subset=["company_name"], keep=False)]
unique_dup_groups = duplicates["company_name"].nunique()
result.affected_records = len(duplicates)
result.passed = len(duplicates) == 0
result.details = (
f"{len(duplicates)} duplicate records in {unique_dup_groups} groups"
if not result.passed else "No duplicates detected"
)
return resultWhat do these three tests check?
test_mdm_completeness() verifies that every MDM golden record has both a sap_bp_number and a salesforce_account_id. A missing reference means the MDM system lost track of which source records belong to that customer.
test_data_freshness() flags SAP records not updated in the last 180 days. Stale data in an ERP often means the sync pipeline stopped pushing updates.
test_duplicate_detection() finds SAP records sharing the same company name under different BP numbers. These are the duplicates your data generator deliberately injected.
- Save e2e_tests.py.
You won't see output yet. The next chunk adds the runner function that executes all five tests and produces the report.
Execute the E2E test suite and generate a structured test report
The runner function calls each test, collects results into a structured JSON report, and writes it to disk. This report format is what you would hand to a stakeholder or feed into a dashboard.
- Add the run_all_e2e_tests() function and the __main__ block at the bottom of e2e_tests.py:
def run_all_e2e_tests() -> list[dict]:
"""Execute all E2E test scenarios."""
RESULTS_DIR.mkdir(exist_ok=True)
tests = [
test_sap_to_salesforce_mapping,
test_field_transformation_accuracy,
test_mdm_completeness,
test_data_freshness,
test_duplicate_detection,
]
results = []
for test_fn in tests:
test_result = test_fn()
results.append(test_result.to_dict())
# Build the structured report
report = {
"test_run_timestamp": datetime.now().isoformat(),
"total_tests": len(results),
"passed": sum(1 for r in results if r["passed"]),
"failed": sum(1 for r in results if not r["passed"]),
"results": results
}
with open(RESULTS_DIR / "e2e_test_report.json", "w") as f:
json.dump(report, f, indent=2)
print(f"\nE2E Test Report: {report['passed']}/{report['total_tests']} passed")
for r in results:
status = "PASS" if r["passed"] else "FAIL"
print(f" [{status}] {r['name']}: {r['details']}")
return results
if __name__ == "__main__":
run_all_e2e_tests()What does this code do?
run_all_e2e_tests() iterates through all five test functions, calls each one, and collects the dictionary results. It builds a report object containing the timestamp, pass/fail counts, and the full results array.
The report is saved to results/e2e_test_report.json so downstream tools (like the Streamlit dashboard you will build later) can read it. The console output gives you immediate feedback on which tests passed or failed.
- Save e2e_tests.py.
- Run the E2E test suite by executing this command in your terminal:
python e2e_tests.pyYou should see output showing the pass/fail status for each of the five tests. Some tests will show FAIL with details about how many records were affected. This is expected because your synthetic data has intentional anomalies injected.
What should I see?
Your terminal output will look something like this:
- The SAP to Salesforce Mapping test will likely FAIL because duplicate records were injected with new BP numbers that have no Salesforce match.
- The Field Transformation Accuracy test will FAIL because the data generator injected inconsistent values (appending " Inc." to some account names).
- The Data Freshness test will FAIL because some records were backdated by 1-2 years.
- The Duplicate Detection test will FAIL because 2% of records were explicitly duplicated.
- Confirm that results/e2e_test_report.json was created by running:
ls results/You should see both validation_results.json (from the previous step) and e2e_test_report.json in the output.
Seeing a FileNotFoundError?
This means the script cannot find your CSV files. Make sure you are running the command from your project root directory (the folder containing the data/ folder).
If the data/ folder is missing entirely, re-run python data_generator.py first to regenerate the synthetic datasets.
Help me debug my E2E test FileNotFoundError
✔️ Awesome, I've got everything!
Great. Double check you have saved e2e_tests.py and that results/e2e_test_report.json exists.
ⓧ I'd like to double check the full code
Here is the complete e2e_tests.py file. Compare your code against this:
"""End-to-end integration test scenarios for cross-system data flows."""
import pandas as pd
from pathlib import Path
from datetime import datetime
import json
DATA_DIR = Path("data")
RESULTS_DIR = Path("results")
class E2ETestResult:
"""Container for a single E2E test result."""
def __init__(self, name: str, scenario: str):
self.name = name
self.scenario = scenario
self.passed = False
self.details = ""
self.affected_records = 0
self.timestamp = datetime.now().isoformat()
def to_dict(self) -> dict:
return {
"name": self.name,
"scenario": self.scenario,
"passed": self.passed,
"details": self.details,
"affected_records": self.affected_records,
"timestamp": self.timestamp
}
def test_sap_to_salesforce_mapping() -> E2ETestResult:
"""Test that SAP records correctly map to Salesforce via external_id."""
result = E2ETestResult(
name="SAP to Salesforce Mapping",
scenario="Verify all SAP BP numbers have corresponding Salesforce accounts"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sf_df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
sap_active = sap_df[sap_df["is_active"] == True]
mapped = sap_active.merge(sf_df, left_on="bp_number", right_on="external_id", how="left")
unmapped = mapped[mapped["account_id"].isna()]
result.affected_records = len(unmapped)
result.passed = len(unmapped) == 0
result.details = (
f"{len(unmapped)} active SAP records have no Salesforce account mapping"
if not result.passed else "All active SAP records mapped to Salesforce"
)
return result
def test_field_transformation_accuracy() -> E2ETestResult:
"""Test that field values are correctly transformed between systems."""
result = E2ETestResult(
name="Field Transformation Accuracy",
scenario="Verify field values match after SAP->Salesforce transformation"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sf_df = pd.read_csv(DATA_DIR / "salesforce_accounts.csv")
merged = sap_df.merge(sf_df, left_on="bp_number", right_on="external_id", how="inner")
field_mappings = [
("company_name", "account_name"),
("street", "billing_street"),
("city", "billing_city"),
("postal_code", "billing_postal_code"),
]
mismatches = 0
mismatch_details = []
for sap_field, sf_field in field_mappings:
diff = merged[merged[sap_field] != merged[sf_field]]
if len(diff) > 0:
mismatches += len(diff)
mismatch_details.append(f"{sap_field}->{sf_field}: {len(diff)} mismatches")
result.affected_records = mismatches
result.passed = mismatches == 0
result.details = (
"; ".join(mismatch_details) if mismatch_details
else "All field transformations accurate"
)
return result
def test_mdm_completeness() -> E2ETestResult:
"""Test that MDM golden records have complete cross-references."""
result = E2ETestResult(
name="MDM Cross-Reference Completeness",
scenario="Verify MDM records have both SAP and Salesforce references"
)
mdm_df = pd.read_csv(DATA_DIR / "mdm_golden_records.csv")
missing_sap = mdm_df[
(mdm_df["sap_bp_number"].isna()) | (mdm_df["sap_bp_number"] == "")
]
missing_sf = mdm_df[
(mdm_df["salesforce_account_id"].isna()) | (mdm_df["salesforce_account_id"] == "")
]
total_incomplete = len(missing_sap) + len(missing_sf)
result.affected_records = total_incomplete
result.passed = total_incomplete == 0
result.details = (
f"Missing SAP refs: {len(missing_sap)}, Missing SF refs: {len(missing_sf)}"
if not result.passed else "All MDM records have complete cross-references"
)
return result
def test_data_freshness() -> E2ETestResult:
"""Test that records are not stale (modified within last 180 days)."""
result = E2ETestResult(
name="Data Freshness",
scenario="Verify records have been updated within the last 180 days"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
sap_df["last_modified"] = pd.to_datetime(sap_df["last_modified"])
threshold = datetime.now() - pd.Timedelta(days=180)
stale = sap_df[sap_df["last_modified"] < threshold]
result.affected_records = len(stale)
result.passed = len(stale) == 0
result.details = (
f"{len(stale)} records not updated in 180+ days"
if not result.passed else "All records are fresh"
)
return result
def test_duplicate_detection() -> E2ETestResult:
"""Test for duplicate records based on company name similarity."""
result = E2ETestResult(
name="Duplicate Detection",
scenario="Identify records with identical company names across different BP numbers"
)
sap_df = pd.read_csv(DATA_DIR / "sap_business_partners.csv")
duplicates = sap_df[sap_df.duplicated(subset=["company_name"], keep=False)]
unique_dup_groups = duplicates["company_name"].nunique()
result.affected_records = len(duplicates)
result.passed = len(duplicates) == 0
result.details = (
f"{len(duplicates)} duplicate records in {unique_dup_groups} groups"
if not result.passed else "No duplicates detected"
)
return result
def run_all_e2e_tests() -> list[dict]:
"""Execute all E2E test scenarios."""
RESULTS_DIR.mkdir(exist_ok=True)
tests = [
test_sap_to_salesforce_mapping,
test_field_transformation_accuracy,
test_mdm_completeness,
test_data_freshness,
test_duplicate_detection,
]
results = []
for test_fn in tests:
test_result = test_fn()
results.append(test_result.to_dict())
report = {
"test_run_timestamp": datetime.now().isoformat(),
"total_tests": len(results),
"passed": sum(1 for r in results if r["passed"]),
"failed": sum(1 for r in results if not r["passed"]),
"results": results
}
with open(RESULTS_DIR / "e2e_test_report.json", "w") as f:
json.dump(report, f, indent=2)
print(f"\nE2E Test Report: {report['passed']}/{report['total_tests']} passed")
for r in results:
status = "PASS" if r["passed"] else "FAIL"
print(f" [{status}] {r['name']}: {r['details']}")
return results
if __name__ == "__main__":
run_all_e2e_tests()Why do tests fail on purpose?
In real enterprise data engineering, the point of E2E tests is to surface problems. Your synthetic data generator deliberately injected anomalies at an 8% rate. The failing tests prove your test suite works. If every test passed, your tests would be too lenient to catch real-world issues.
Your E2E test suite is now detecting real data quality issues across the SAP, Salesforce, and MDM pipeline. Next, you will classify these detected anomalies by root cause and generate a prioritized action plan for remediation.
Build Root Cause Analysis and Anomaly Classification
Your validation engine and E2E tests now detect data quality problems across SAP, Salesforce, and MDM. Detection alone is not enough for stakeholders, though. They need to understand why problems occur and which ones to fix first.
In this step, you'll build a root cause analysis module that classifies every anomaly by type and severity. It produces a prioritized action plan that tells stakeholders exactly what to fix and in what order.
In this step, get ready to:
- Build an anomaly classifier that maps each data quality issue to a root cause category with severity scoring.
- Generate a structured RCA report with severity breakdowns, system attribution, and a prioritized remediation action plan.
Build the anomaly classifier
Root cause analysis turns raw anomaly detections into actionable intelligence. Instead of saying "80 records failed," you want to say "35 are format mismatches (medium severity), 25 are missing mappings (critical), and 20 are duplicates (medium)."
- Create a new file called root_cause_analysis.py in your project folder.
- Add the imports, path constants, and the anomaly categories dictionary by pasting this code:
"""Root cause analysis module for classifying and reporting data anomalies."""
import pandas as pd
from pathlib import Path
from collections import Counter
import json
DATA_DIR = Path("data")
RESULTS_DIR = Path("results")
ANOMALY_CATEGORIES = {
"mapping_error": {
"description": "Incorrect or missing field mappings between systems",
"severity": "high",
"remediation": "Review integration mapping configuration"
},
"duplicate": {
"description": "Same entity represented by multiple records",
"severity": "medium",
"remediation": "Merge duplicate records using golden record logic"
},
"format_mismatch": {
"description": "Data format inconsistencies (dates, codes, phone numbers)",
"severity": "medium",
"remediation": "Apply format standardization rules"
},
"missing_field": {
"description": "Required fields are empty or null",
"severity": "high",
"remediation": "Enrich from alternate source system or flag for manual entry"
},
"stale_record": {
"description": "Record not updated within expected freshness window",
"severity": "low",
"remediation": "Trigger re-sync from source system"
},
"inconsistent_value": {
"description": "Same field has different values across systems",
"severity": "high",
"remediation": "Apply source priority rules to determine canonical value"
},
"missing_mapping": {
"description": "Cross-reference link between systems is broken or absent",
"severity": "critical",
"remediation": "Re-establish system linkage via matching algorithm"
}
}What does this code do?
The ANOMALY_CATEGORIES dictionary is the knowledge base of your classifier. Each key maps to a known anomaly pattern your data generator injected in Step 2.
- Each category has a description (what it means), a severity level (critical/high/medium/low), and a remediation action (what to do about it).
- Severity scoring lets stakeholders triage. A missing_mapping is critical because broken cross-references mean records are invisible across systems. A stale_record is low because the data exists but is outdated.
- Save root_cause_analysis.py.
Now you'll add the function that reads your anomaly log and classifies each issue against the categories dictionary.
- In root_cause_analysis.py, add the classify_anomalies() function below the ANOMALY_CATEGORIES dictionary:
def classify_anomalies() -> dict:
"""Load anomaly log and classify issues by root cause."""
anomaly_df = pd.read_csv(DATA_DIR / "anomaly_log.csv")
classification = {
"total_anomalies": len(anomaly_df),
"by_type": {},
"by_system": {},
"by_severity": Counter(),
"details": []
}
# Count anomalies by type and enrich with category metadata
type_counts = anomaly_df["anomaly_type"].value_counts().to_dict()
for anomaly_type, count in type_counts.items():
category = ANOMALY_CATEGORIES.get(anomaly_type, {
"description": "Unknown anomaly type",
"severity": "medium",
"remediation": "Manual investigation required"
})
classification["by_type"][anomaly_type] = {
"count": count,
"percentage": round(count / len(anomaly_df) * 100, 1),
**category
}
classification["by_severity"][category["severity"]] += count
# Count anomalies by source system
system_counts = anomaly_df["system"].value_counts().to_dict()
for system, count in system_counts.items():
classification["by_system"][system] = {
"count": count,
"percentage": round(count / len(anomaly_df) * 100, 1)
}
# Build per-record detail list with severity and remediation
for _, row in anomaly_df.iterrows():
category = ANOMALY_CATEGORIES.get(row["anomaly_type"], {})
classification["details"].append({
"record_index": int(row["record_index"]),
"system": row["system"],
"anomaly_type": row["anomaly_type"],
"severity": category.get("severity", "medium"),
"remediation": category.get("remediation", "Manual review"),
"bp_number": row["bp_number"]
})
return classificationWhat does this function do?
- It reads anomaly_log.csv (the log your data generator created in Step 2) and groups anomalies three ways: by type, by system, and by severity.
- Each anomaly type gets enriched with its metadata from ANOMALY_CATEGORIES using the spread operator (**category).
- The Counter() accumulates severity totals so you can quickly see how many critical vs. low issues exist.
- The details list preserves per-record granularity for drill-down views in the dashboard later.
- Save root_cause_analysis.py.
Seeing a KeyError or empty classification?
- Make sure your data/anomaly_log.csv file exists. If it doesn't, re-run data_generator.py from Step 2.
- Check that the column names in the CSV match what the code expects: anomaly_type, system, record_index, and bp_number.
- Help me debug my classify_anomalies function
Generate the root cause analysis report
The classifier organizes raw data. Now you need a report generator that structures those classifications into a stakeholder-ready format with a prioritized action plan.
- In root_cause_analysis.py, add the generate_rca_report() function and the entry point below classify_anomalies():
def generate_rca_report() -> dict:
"""Generate a comprehensive root cause analysis report."""
RESULTS_DIR.mkdir(exist_ok=True)
classification = classify_anomalies()
report = {
"report_title": "Data Quality Root Cause Analysis",
"generated_at": pd.Timestamp.now().isoformat(),
"summary": {
"total_anomalies": classification["total_anomalies"],
"critical": classification["by_severity"].get("critical", 0),
"high": classification["by_severity"].get("high", 0),
"medium": classification["by_severity"].get("medium", 0),
"low": classification["by_severity"].get("low", 0),
},
"by_type": classification["by_type"],
"by_system": classification["by_system"],
"action_plan": [],
"details": classification["details"]
}
# Generate prioritized action plan
sorted_types = sorted(
classification["by_type"].items(),
key=lambda x: {"critical": 0, "high": 1, "medium": 2, "low": 3}.get(
x[1]["severity"], 4
)
)
for i, (anomaly_type, info) in enumerate(sorted_types, 1):
report["action_plan"].append({
"priority": i,
"issue": anomaly_type,
"severity": info["severity"],
"affected_records": info["count"],
"action": info["remediation"]
})
with open(RESULTS_DIR / "rca_report.json", "w") as f:
json.dump(report, f, indent=2, default=str)
print(f"\nRoot Cause Analysis Report Generated")
print(f" Total anomalies: {report['summary']['total_anomalies']}")
print(f" Critical: {report['summary']['critical']}")
print(f" High: {report['summary']['high']}")
print(f" Medium: {report['summary']['medium']}")
print(f" Low: {report['summary']['low']}")
print(f"\n Action Plan:")
for action in report["action_plan"]:
print(f" {action['priority']}. [{action['severity'].upper()}] {action['issue']}: {action['action']}")
return report
if __name__ == "__main__":
generate_rca_report()What does this function do?
- It calls classify_anomalies() to get the raw classification, then wraps it into a structured report with a title, timestamp, and summary counts.
- The action plan sorts anomaly types by severity (critical first, low last) so stakeholders see the most urgent issues at the top.
- The report is saved to results/rca_report.json and also printed to the terminal for quick review.
- Save root_cause_analysis.py.
- Run the root cause analysis report by executing this command in your terminal:
python root_cause_analysis.pyYou should see output listing the total anomalies, counts by severity level, and a numbered action plan sorted from critical to low priority.
- Confirm that results/rca_report.json was created by running:
ls results/You should see rca_report.json alongside validation_results.json and e2e_test_report.json from earlier steps.
Report shows zero anomalies?
- Open data/anomaly_log.csv and verify it has rows. If it's empty, your data generator may not have injected anomalies. Re-run python data_generator.py to regenerate.
- Check that DATA_DIR points to Path("data") and you're running the script from your project folder (not a subfolder).
- Help me troubleshoot my RCA report generation
Why prioritize by severity instead of count?
A single critical-severity issue (like a broken cross-reference) can make an entire customer invisible across all systems. Ten low-severity stale records are annoying but don't block operations. Stakeholders need to allocate engineering time to the issues with the highest business impact, not just the highest volume.
✔️ 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 the complete root_cause_analysis.py file for comparison:
"""Root cause analysis module for classifying and reporting data anomalies."""
import pandas as pd
from pathlib import Path
from collections import Counter
import json
DATA_DIR = Path("data")
RESULTS_DIR = Path("results")
ANOMALY_CATEGORIES = {
"mapping_error": {
"description": "Incorrect or missing field mappings between systems",
"severity": "high",
"remediation": "Review integration mapping configuration"
},
"duplicate": {
"description": "Same entity represented by multiple records",
"severity": "medium",
"remediation": "Merge duplicate records using golden record logic"
},
"format_mismatch": {
"description": "Data format inconsistencies (dates, codes, phone numbers)",
"severity": "medium",
"remediation": "Apply format standardization rules"
},
"missing_field": {
"description": "Required fields are empty or null",
"severity": "high",
"remediation": "Enrich from alternate source system or flag for manual entry"
},
"stale_record": {
"description": "Record not updated within expected freshness window",
"severity": "low",
"remediation": "Trigger re-sync from source system"
},
"inconsistent_value": {
"description": "Same field has different values across systems",
"severity": "high",
"remediation": "Apply source priority rules to determine canonical value"
},
"missing_mapping": {
"description": "Cross-reference link between systems is broken or absent",
"severity": "critical",
"remediation": "Re-establish system linkage via matching algorithm"
}
}
def classify_anomalies() -> dict:
"""Load anomaly log and classify issues by root cause."""
anomaly_df = pd.read_csv(DATA_DIR / "anomaly_log.csv")
classification = {
"total_anomalies": len(anomaly_df),
"by_type": {},
"by_system": {},
"by_severity": Counter(),
"details": []
}
type_counts = anomaly_df["anomaly_type"].value_counts().to_dict()
for anomaly_type, count in type_counts.items():
category = ANOMALY_CATEGORIES.get(anomaly_type, {
"description": "Unknown anomaly type",
"severity": "medium",
"remediation": "Manual investigation required"
})
classification["by_type"][anomaly_type] = {
"count": count,
"percentage": round(count / len(anomaly_df) * 100, 1),
**category
}
classification["by_severity"][category["severity"]] += count
system_counts = anomaly_df["system"].value_counts().to_dict()
for system, count in system_counts.items():
classification["by_system"][system] = {
"count": count,
"percentage": round(count / len(anomaly_df) * 100, 1)
}
for _, row in anomaly_df.iterrows():
category = ANOMALY_CATEGORIES.get(row["anomaly_type"], {})
classification["details"].append({
"record_index": int(row["record_index"]),
"system": row["system"],
"anomaly_type": row["anomaly_type"],
"severity": category.get("severity", "medium"),
"remediation": category.get("remediation", "Manual review"),
"bp_number": row["bp_number"]
})
return classification
def generate_rca_report() -> dict:
"""Generate a comprehensive root cause analysis report."""
RESULTS_DIR.mkdir(exist_ok=True)
classification = classify_anomalies()
report = {
"report_title": "Data Quality Root Cause Analysis",
"generated_at": pd.Timestamp.now().isoformat(),
"summary": {
"total_anomalies": classification["total_anomalies"],
"critical": classification["by_severity"].get("critical", 0),
"high": classification["by_severity"].get("high", 0),
"medium": classification["by_severity"].get("medium", 0),
"low": classification["by_severity"].get("low", 0),
},
"by_type": classification["by_type"],
"by_system": classification["by_system"],
"action_plan": [],
"details": classification["details"]
}
# Generate prioritized action plan
sorted_types = sorted(
classification["by_type"].items(),
key=lambda x: {"critical": 0, "high": 1, "medium": 2, "low": 3}.get(
x[1]["severity"], 4
)
)
for i, (anomaly_type, info) in enumerate(sorted_types, 1):
report["action_plan"].append({
"priority": i,
"issue": anomaly_type,
"severity": info["severity"],
"affected_records": info["count"],
"action": info["remediation"]
})
with open(RESULTS_DIR / "rca_report.json", "w") as f:
json.dump(report, f, indent=2, default=str)
print(f"\nRoot Cause Analysis Report Generated")
print(f" Total anomalies: {report['summary']['total_anomalies']}")
print(f" Critical: {report['summary']['critical']}")
print(f" High: {report['summary']['high']}")
print(f" Medium: {report['summary']['medium']}")
print(f" Low: {report['summary']['low']}")
print(f"\n Action Plan:")
for action in report["action_plan"]:
print(f" {action['priority']}. [{action['severity'].upper()}] {action['issue']}: {action['action']}")
return report
if __name__ == "__main__":
generate_rca_report()Your root cause analysis module now transforms raw anomaly detections into a structured, prioritized report. Next up, you'll bring all of this data to life in a Streamlit dashboard that stakeholders can actually explore.
Build the Stakeholder Data Quality Dashboard
Your validation engine, E2E tests, and root cause analysis module all produce JSON reports. These reports are powerful for engineers, but stakeholders need something visual and interactive.
In this step, you will build a Streamlit dashboard that reads those report files and presents them as quality scores, test result cards, and filterable anomaly tables.
In this step, get ready to:
- Create a multi-page Streamlit app with sidebar navigation.
- Display quality metrics with progress bars for each system.
- Build interactive anomaly drill-down views with filtering.
Create the dashboard layout and navigation
Streamlit apps are plain Python scripts that Streamlit re-runs top-to-bottom whenever a user interacts with a widget. You will define a main() entry point that configures the page, loads your result files, and routes users to one of three views via a sidebar radio button.
- Create a new file called dashboard.py in your data-quality-engine project folder.
- Add the imports, results directory constant, and the load_results() function by pasting this code:
"""Streamlit dashboard for data quality monitoring and stakeholder reporting."""
import streamlit as st
import pandas as pd
import json
from pathlib import Path
RESULTS_DIR = Path("results")
def load_results():
# Load all JSON result files into a single dictionary
data = {}
validation_path = RESULTS_DIR / "validation_results.json"
e2e_path = RESULTS_DIR / "e2e_test_report.json"
rca_path = RESULTS_DIR / "rca_report.json"
if validation_path.exists():
with open(validation_path) as f:
data["validation"] = json.load(f)
if e2e_path.exists():
with open(e2e_path) as f:
data["e2e"] = json.load(f)
if rca_path.exists():
with open(rca_path) as f:
data["rca"] = json.load(f)
return dataWhat does this code do?
RESULTS_DIR points to the results/ folder where your previous scripts saved their JSON output.
load_results() checks if each report file exists and loads it into a dictionary. The existence checks prevent crashes if you run the dashboard before generating all reports.
- Below load_results(), add the main() function and entry point that wires up the page config and sidebar navigation:
def main():
# Configure the Streamlit page settings
st.set_page_config(page_title="Data Quality Dashboard", page_icon="\ud83d\udcca", layout="wide")
st.title("\ud83d\udcca Enterprise Data Quality Dashboard")
st.caption("Cross-system data validation for SAP, Salesforce, and MDM")
# Load all result data from JSON files
data = load_results()
# Sidebar navigation between three views
page = st.sidebar.radio("Navigation", ["Overview", "E2E Test Results", "Anomaly Drill-Down"])
if page == "Overview":
render_overview(data)
elif page == "E2E Test Results":
render_test_results(data)
elif page == "Anomaly Drill-Down":
render_anomaly_drilldown(data)
if __name__ == "__main__":
main()What does this code do?
st.set_page_config() sets the browser tab title and uses a wide layout for more space. This must be the first Streamlit command in the script.
st.sidebar.radio() creates a set of radio buttons in the left sidebar. The selected page name gets stored in page, and the if/elif block calls the matching render function.
Implement the quality metrics overview
The Overview page gives stakeholders a single glance at how healthy the data is. It calculates an overall quality score from validation results and shows per-system progress bars.
- Add the render_overview() function above main() in dashboard.py:
def render_overview(data: dict):
st.header("Data Quality Overview")
if "validation" not in data:
st.warning("No validation results found. Run validation_engine.py first.")
return
# Calculate overall quality score from all validation checks
total_passed = sum(r["passed"] for r in data["validation"])
total_checks = sum(r["total"] for r in data["validation"])
overall_score = (total_passed / total_checks * 100) if total_checks > 0 else 0
# Display top-level metrics in three columns
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Overall Quality Score", f"{overall_score:.1f}%")
with col2:
st.metric("Checks Passed", f"{total_passed}/{total_checks}")
with col3:
e2e_passed = data.get("e2e", {}).get("passed", 0)
e2e_total = data.get("e2e", {}).get("total_tests", 0)
st.metric("E2E Tests Passed", f"{e2e_passed}/{e2e_total}")
st.divider()
# Show per-system quality as progress bars
st.subheader("Quality by System")
for result in data["validation"]:
system = result["system"]
passed = result["passed"]
total = result["total"]
score = (passed / total * 100) if total > 0 else 0
col_a, col_b = st.columns([3, 1])
with col_a:
st.progress(score / 100, text=f"{system}: {score:.0f}%")
with col_b:
st.write(f"{passed}/{total} passed")What does this code do?
st.metric() renders big bold numbers that stakeholders can read at a glance. The three columns put the overall score, checks passed, and E2E results side by side.
st.progress() takes a float between 0.0 and 1.0 and renders a horizontal progress bar. The text parameter shows a label above the bar identifying the system and its percentage.
Add the anomaly drill-down and test progress views
Stakeholders also need to drill into specific test failures and anomaly details. You will add two more render functions: one for E2E test results with expandable cards, and one for the full anomaly breakdown with interactive filters.
- Add the render_test_results() function above main() in dashboard.py:
def render_test_results(data: dict):
st.header("E2E Integration Test Results")
if "e2e" not in data:
st.warning("No E2E test results found. Run e2e_tests.py first.")
return
e2e = data["e2e"]
st.write(f"**Test Run**: {e2e['test_run_timestamp']}")
st.write(f"**Results**: {e2e['passed']} passed, {e2e['failed']} failed out of {e2e['total_tests']} tests")
st.divider()
# Render each test as an expandable card
for test in e2e["results"]:
status_icon = "\u2705" if test["passed"] else "\u274c"
with st.expander(f"{status_icon} {test['name']}"):
st.write(f"**Scenario**: {test['scenario']}")
st.write(f"**Status**: {'Passed' if test['passed'] else 'Failed'}")
st.write(f"**Details**: {test['details']}")
st.write(f"**Affected Records**: {test['affected_records']}")What does this code do?
st.expander() creates a collapsible section for each test. Stakeholders can expand only the failing tests to read the details without scrolling through passing ones.
The status icon is a green checkmark for passing tests and a red cross for failures, giving instant visual feedback.
- Add the render_anomaly_drilldown() function above main() in dashboard.py:
def render_anomaly_drilldown(data: dict):
st.header("Anomaly Drill-Down")
if "rca" not in data:
st.warning("No RCA report found. Run root_cause_analysis.py first.")
return
rca = data["rca"]
# Summary metrics in four columns
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Total Anomalies", rca["summary"]["total_anomalies"])
with col2:
st.metric("Critical", rca["summary"]["critical"])
with col3:
st.metric("High", rca["summary"]["high"])
with col4:
st.metric("Medium + Low", rca["summary"]["medium"] + rca["summary"]["low"])
st.divider()
# Anomaly type breakdown table
st.subheader("Anomalies by Type")
type_data = []
for atype, info in rca["by_type"].items():
type_data.append({"Type": atype, "Count": info["count"], "Severity": info["severity"], "Percentage": f"{info['percentage']}%", "Remediation": info["remediation"]})
if type_data:
st.dataframe(pd.DataFrame(type_data), use_container_width=True)
# System breakdown table
st.subheader("Anomalies by System")
system_data = []
for system, info in rca["by_system"].items():
system_data.append({"System": system, "Count": info["count"], "Percentage": f"{info['percentage']}%"})
if system_data:
st.dataframe(pd.DataFrame(system_data), use_container_width=True)
# Prioritized action plan
st.subheader("Recommended Action Plan")
for action in rca["action_plan"]:
severity_color = {"critical": "\ud83d\udd34", "high": "\ud83d\udfe0", "medium": "\ud83d\udfe1", "low": "\ud83d\udfe2"}.get(action["severity"], "\u26aa")
st.write(f"{action['priority']}. {severity_color} **{action['issue']}** ({action['affected_records']} records) - {action['action']}")
st.divider()
# Filterable detailed anomaly records
st.subheader("Detailed Anomaly Records")
details_df = pd.DataFrame(rca["details"])
if not details_df.empty:
col_filter1, col_filter2 = st.columns(2)
with col_filter1:
type_filter = st.multiselect("Filter by Type", options=details_df["anomaly_type"].unique(), default=details_df["anomaly_type"].unique())
with col_filter2:
system_filter = st.multiselect("Filter by System", options=details_df["system"].unique(), default=details_df["system"].unique())
filtered = details_df[(details_df["anomaly_type"].isin(type_filter)) & (details_df["system"].isin(system_filter))]
st.dataframe(filtered, use_container_width=True)What does this code do?
The top metrics row shows severity counts at a glance. Below that, two st.dataframe() calls render the anomaly breakdown by type and by system as interactive tables that stakeholders can sort.
The action plan section lists remediation steps in priority order with color-coded severity indicators. The "Detailed Anomaly Records" section uses st.multiselect() widgets to let users filter the full anomaly table by type and system.
- Save dashboard.py.
- Launch the dashboard by running this command in your terminal:
streamlit run dashboard.pyYour browser opens automatically to http://localhost:8501. You should see the Enterprise Data Quality Dashboard title, a sidebar with three navigation options, and the Overview page showing quality scores and per-system progress bars.
- Click E2E Test Results in the sidebar to see expandable test cards.
- Click Anomaly Drill-Down to see the metrics, tables, action plan, and filterable records.
Dashboard not loading?
Make sure you saved dashboard.py before running the command. Streamlit only reads from the saved file on disk.
If you see a NameError about render_overview not being defined, check that all four functions (load_results, render_overview, render_test_results, render_anomaly_drilldown) appear above main() in the file.
help me debug my Streamlit dashboard
✔️ Awesome, I've got everything!
Great. Double check you have saved dashboard.py and that the Streamlit server is still running in your terminal.
ⓧ I'd like to double check the full code
"""Streamlit dashboard for data quality monitoring and stakeholder reporting."""
import streamlit as st
import pandas as pd
import json
from pathlib import Path
RESULTS_DIR = Path("results")
def load_results():
data = {}
validation_path = RESULTS_DIR / "validation_results.json"
e2e_path = RESULTS_DIR / "e2e_test_report.json"
rca_path = RESULTS_DIR / "rca_report.json"
if validation_path.exists():
with open(validation_path) as f:
data["validation"] = json.load(f)
if e2e_path.exists():
with open(e2e_path) as f:
data["e2e"] = json.load(f)
if rca_path.exists():
with open(rca_path) as f:
data["rca"] = json.load(f)
return data
def render_overview(data: dict):
st.header("Data Quality Overview")
if "validation" not in data:
st.warning("No validation results found. Run validation_engine.py first.")
return
total_passed = sum(r["passed"] for r in data["validation"])
total_checks = sum(r["total"] for r in data["validation"])
overall_score = (total_passed / total_checks * 100) if total_checks > 0 else 0
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Overall Quality Score", f"{overall_score:.1f}%")
with col2:
st.metric("Checks Passed", f"{total_passed}/{total_checks}")
with col3:
e2e_passed = data.get("e2e", {}).get("passed", 0)
e2e_total = data.get("e2e", {}).get("total_tests", 0)
st.metric("E2E Tests Passed", f"{e2e_passed}/{e2e_total}")
st.divider()
st.subheader("Quality by System")
for result in data["validation"]:
system = result["system"]
passed = result["passed"]
total = result["total"]
score = (passed / total * 100) if total > 0 else 0
col_a, col_b = st.columns([3, 1])
with col_a:
st.progress(score / 100, text=f"{system}: {score:.0f}%")
with col_b:
st.write(f"{passed}/{total} passed")
def render_test_results(data: dict):
st.header("E2E Integration Test Results")
if "e2e" not in data:
st.warning("No E2E test results found. Run e2e_tests.py first.")
return
e2e = data["e2e"]
st.write(f"**Test Run**: {e2e['test_run_timestamp']}")
st.write(f"**Results**: {e2e['passed']} passed, {e2e['failed']} failed out of {e2e['total_tests']} tests")
st.divider()
for test in e2e["results"]:
status_icon = "\u2705" if test["passed"] else "\u274c"
with st.expander(f"{status_icon} {test['name']}"):
st.write(f"**Scenario**: {test['scenario']}")
st.write(f"**Status**: {'Passed' if test['passed'] else 'Failed'}")
st.write(f"**Details**: {test['details']}")
st.write(f"**Affected Records**: {test['affected_records']}")
def render_anomaly_drilldown(data: dict):
st.header("Anomaly Drill-Down")
if "rca" not in data:
st.warning("No RCA report found. Run root_cause_analysis.py first.")
return
rca = data["rca"]
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Total Anomalies", rca["summary"]["total_anomalies"])
with col2:
st.metric("Critical", rca["summary"]["critical"])
with col3:
st.metric("High", rca["summary"]["high"])
with col4:
st.metric("Medium + Low", rca["summary"]["medium"] + rca["summary"]["low"])
st.divider()
st.subheader("Anomalies by Type")
type_data = []
for atype, info in rca["by_type"].items():
type_data.append({"Type": atype, "Count": info["count"], "Severity": info["severity"], "Percentage": f"{info['percentage']}%", "Remediation": info["remediation"]})
if type_data:
st.dataframe(pd.DataFrame(type_data), use_container_width=True)
st.subheader("Anomalies by System")
system_data = []
for system, info in rca["by_system"].items():
system_data.append({"System": system, "Count": info["count"], "Percentage": f"{info['percentage']}%"})
if system_data:
st.dataframe(pd.DataFrame(system_data), use_container_width=True)
st.subheader("Recommended Action Plan")
for action in rca["action_plan"]:
severity_color = {"critical": "\ud83d\udd34", "high": "\ud83d\udfe0", "medium": "\ud83d\udfe1", "low": "\ud83d\udfe2"}.get(action["severity"], "\u26aa")
st.write(f"{action['priority']}. {severity_color} **{action['issue']}** ({action['affected_records']} records) - {action['action']}")
st.divider()
st.subheader("Detailed Anomaly Records")
details_df = pd.DataFrame(rca["details"])
if not details_df.empty:
col_filter1, col_filter2 = st.columns(2)
with col_filter1:
type_filter = st.multiselect("Filter by Type", options=details_df["anomaly_type"].unique(), default=details_df["anomaly_type"].unique())
with col_filter2:
system_filter = st.multiselect("Filter by System", options=details_df["system"].unique(), default=details_df["system"].unique())
filtered = details_df[(details_df["anomaly_type"].isin(type_filter)) & (details_df["system"].isin(system_filter))]
st.dataframe(filtered, use_container_width=True)
def main():
st.set_page_config(page_title="Data Quality Dashboard", page_icon="\ud83d\udcca", layout="wide")
st.title("\ud83d\udcca Enterprise Data Quality Dashboard")
st.caption("Cross-system data validation for SAP, Salesforce, and MDM")
data = load_results()
page = st.sidebar.radio("Navigation", ["Overview", "E2E Test Results", "Anomaly Drill-Down"])
if page == "Overview":
render_overview(data)
elif page == "E2E Test Results":
render_test_results(data)
elif page == "Anomaly Drill-Down":
render_anomaly_drilldown(data)
if __name__ == "__main__":
main()Secret mission
Automated Anomaly Remediation Engine
Your validation engine detects problems, but what if it could fix them too? Build an automated remediation system that proposes cleanup actions, executes them in dry-run mode, and logs every change to a SQLite audit trail for rollback capability.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely on your local machine, so there are no ongoing cloud costs or subscription charges. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Streamlit development server (if still running in your terminal).
- Project folder on your Desktop containing all Python files, the data/ directory (CSV files), the results/ directory (JSON reports and SQLite database), the virtual environment, and GX context files.
Keep everything running
No action needed. Choose this if you plan to keep experimenting with the validation engine, adding new test scenarios, or extending the remediation rules.
- Your Streamlit dashboard stays accessible at http://localhost:8501 as long as the terminal process is running.
- All generated data, validation results, and the remediation audit database remain intact in your project folder.
- Python 3.12 and VS Code remain installed on your system.
Pause - I'll come back to this later
Shut down running processes to free up system resources. Your project files stay intact so you can pick up where you left off.
- In the terminal where Streamlit is running, press Ctrl+C to stop the development server.
- Deactivate your virtual environment by running:
deactivate- To resume later, navigate back to your project folder, reactivate the virtual environment, and run streamlit run dashboard.py.
Delete - I don't want to use this again
Remove all project files and start fresh. This permanently deletes your generated data, validation results, remediation audit database, and all Python source files.
- If Streamlit is still running, press Ctrl+C in the terminal to stop it.
- Deactivate your virtual environment by running:
deactivate- Navigate to your Desktop and delete the project folder by running:
cd ~/Desktop
rm -rf data-quality-engineWindows users
If you are on Windows, use this command in PowerShell instead:
cd ~\Desktop
Remove-Item -Recurse -Force data-quality-engine- Confirm the folder is gone by running:
ls- You should no longer see the project folder in the output.
- Python 3.12 and VS Code remain installed on your system. These are general-purpose tools useful for future projects.
Nice Work!
Nice Work!
You just built an enterprise-grade data quality validation platform from scratch. This system simulates and tests real-world data flows between SAP, Salesforce, and an MDM system, catching the exact anomalies that plague multi-system integrations.
You've learned how to:
- Design Pydantic data contracts and generate synthetic enterprise data with controlled anomaly injection across SAP, Salesforce, and MDM systems.
- Build cross-system validation suites with Great Expectations and E2E integration tests that verify data transformation accuracy, referential integrity, and format consistency.
- Classify anomalies with a root cause analysis engine and present quality metrics through a professional Streamlit dashboard for stakeholder communication.
- Built an automated remediation engine with dry-run capability, fuzzy-match deduplication, cross-system field enrichment, and a SQLite audit log for full rollback traceability.
Ready to quiz yourself?