Validate Data completeness

Question

Currently all database data is migrated from SQL/Oracle to snowflake( i.e. Extraction and Load only), basically a replica of Data so that later we can modify the data and Power Bi can fetch from Snowflake.

For now my requirement is to just validate if row and column count in tables between SQL and Snowflake is same i.e. validate data completeness for thousands or millions of records without a paid tool.

Since I have around 1000+Tables in database, I'm okay to run a python script to do the validation. Or Please suggest any other input that could help in achieving this.

Answer 1

To validate the completeness of the data in thousands or millions of records between SQL and Snowflake, you can write a Python script to compare the row and column count of tables in both databases. Here are the steps to achieve this:

1. Connect to the SQL and Snowflake databases using Python libraries like pyodbc and snowflake-connector-python.

2. Get the list of table names from both databases using the SELECT statement.

3. Loop through each table name and fetch the row and column count using the COUNT(*) and COUNT(column_name) functions respectively.

4. Compare the row and column count of each table in both databases.

5. Log the result for each table in a file or a database for future reference.

Here's a sample Python script to get you started:

import pyodbc
import snowflake.connector

# Connect to SQL Server
sql_conn = pyodbc.connect('DRIVER={SQL Server};SERVER=<your server_name here>;DATABASE=<database_name>;UID=<your username here>;PWD=<your password here>')

# Connect to Snowflake
snowflake_conn = snowflake.connector.connect(
    user='<your username here>',
    password='<your password here>',
    account='<your account_name here>',
    warehouse='<your warehouse_name here>',
    database='<your database_name here>',
    schema='<your schema_name here>'
)

# Get the list of table names from SQL Server
sql_cursor = sql_conn.cursor()
sql_cursor.execute('SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = \'BASE TABLE\'')
sql_tables = sql_cursor.fetchall()

# Get the list of table names from Snowflake
snowflake_cursor = snowflake_conn.cursor()
snowflake_cursor.execute('SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = \'BASE TABLE\'')
snowflake_tables = snowflake_cursor.fetchall()

# Compare row and column count of tables in both databases
for table_name in sql_tables:
    # Get row count from SQL Server
    sql_cursor.execute(f'SELECT COUNT(*) FROM {table_name[0]}')
    sql_row_count = sql_cursor.fetchone()[0]
    
    # Get column count from SQL Server
    sql_cursor.execute(f'SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = \'{table_name[0]}\'')
    sql_column_count = sql_cursor.fetchone()[0]
    
    # Get row count from Snowflake
    snowflake_cursor.execute(f'SELECT COUNT(*) FROM {table_name[0]}')
    snowflake_row_count = snowflake_cursor.fetchone()[0]
    
    # Get column count from Snowflake
    snowflake_cursor.execute(f'SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = \'{table_name[0]}\'')
    snowflake_column_count = snowflake_cursor.fetchone()[0]
    
    # Compare row and column count
    if sql_row_count == snowflake_row_count and sql_column_count == snowflake_column_count:
        print(f'{table_name[0]}: Data completeness validated')
    else:
        print(f'{table_name[0]}: Data completeness validation failed')

This is a basic example to validate data completeness between SQL Server and Snowflake. You can modify this script as per your requirement and also use it to log the result in a file or a database.

References:

• pyodbc documentation: https://github.com/mkleehammer/pyodbc/wiki
• snowflake-connector-python documentation: https://docs.snowflake.com/en/user-guide/python-connector.html