Scribd
Upload
10 views3 pages

Python ETL Validation Script Guide

This Python script utilizes the pyodbc library to connect to SQL Server and perform ETL validation checks, generating a console report. It includes tests for record count reconciliation, duplicate records, and data integrity, providing a 'PASS' or 'FAIL' status for each check. The script emphasizes efficiency by leveraging SQL Server's capabilities to handle large datasets without pulling all data into Python.

Uploaded by

maheshtester1919
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd
10 views3 pages

Python ETL Validation Script Guide

This Python script utilizes the pyodbc library to connect to SQL Server and perform ETL validation checks, generating a console report. It includes tests for record count reconciliation, duplicate records, and data integrity, providing a 'PASS' or 'FAIL' status for each check. The script emphasizes efficiency by leveraging SQL Server's capabilities to handle large datasets without pulling all data into Python.

Uploaded by

maheshtester1919
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd

This Python script uses the pyodbc library to connect to SQL Server and run the validation

checks we discussed. It aggregates the results into a clean, console-based report.


Prerequisites

You will need to install the driver and library first:


pip install pyodbc

Python ETL Validation Script

import pyodbc

# 1. Database Configuration
conn_str = (
"Driver={SQL Server};"
"Server=YOUR_SERVER_NAME;"
"Database=YOUR_DATABASE_NAME;"
"Trusted_Connection=yes;" # Use UID/PWD if not using Windows Auth
)

def run_validation():
try:
conn = [Link](conn_str)
cursor = [Link]()
print("--- ETL Validation Report ---\n")

# TEST 1: Record Count Reconciliation


[Link]("""
SELECT
(SELECT COUNT(*) FROM Source_Staging) AS src,
(SELECT COUNT(*) FROM Target_Table) AS tgt
""")
src_count, tgt_count = [Link]()
status = "PASS" if src_count == tgt_count else "FAIL"
print(f"[Count Check] Source: {src_count} | Target: {tgt_count} -> {status}")

# TEST 2: Duplicate Check


[Link]("""
SELECT COUNT(*) FROM (
SELECT EmployeeID FROM Target_Table GROUP BY EmployeeID HAVING COUNT(*)
>1
) AS Dups
""")
dup_count = [Link]()[0]
status = "PASS" if dup_count == 0 else "FAIL"
print(f"[Duplicate Check] Found {dup_count} duplicates -> {status}")

# TEST 3: Data Integrity (Mismatched Rows)


[Link]("""
SELECT COUNT(*) FROM (
SELECT * FROM Source_Staging EXCEPT SELECT * FROM Target_Table
) AS Mismatches
""")
mismatch_count = [Link]()[0]
status = "PASS" if mismatch_count == 0 else "FAIL"
print(f"[Data Match Check] Mismatched rows: {mismatch_count} -> {status}")

[Link]()
print("\n--- Validation Complete ---")

except Exception as e:
print(f"Error during validation: {e}")

if __name__ == "__main__":
run_validation()

How this script works

1. Connectivity: It opens a single connection to your SQL Server instance using your
local credentials.
2. Efficiency: Instead of pulling all data into Python (which would be slow for millions of
rows), it lets SQL Server do the heavy lifting using the EXCEPT and COUNT operators.
3. Reporting: It provides an immediate visual "PASS" or "FAIL" for the three most
common ETL failure points: volume, uniqueness, and accuracy.
Summary of Tests

Test Name SQL Logic Used Purpose


Count Check COUNT(*) comparison Ensures no data was dropped
during the load.
Duplicate Check GROUP BY / HAVING Ensures the ETL didn't double-
insert records.
Data Match EXCEPT Ensures the content of every
column is identical.

You might also like