8 Data Governance Tests Every Data Engineer Should Know

Introduction

Bad data doesn't announce itself. A NULL in a few customer_id fields, a duplicate order number, or a column setup as a string when it should be an integer can all be painful. By the time someone notices, it's already downstream in a dashboard or a financial report. The executives are in panic mode, and you and your team are sweating bullets. However, with the strategies listed below, most of these problems can be prevented.

This article walks through eight test patterns that every data engineer should have in their toolkit. Just by copying these simple code snippets, you can start going down the path of better data governance. Each section shows a simple SQL script to understand the concept. It then provides a Python function you can add to your current pipelines. This quick read will surely save you hours of frustration.

1. Completeness: Check for NULLs in Your Tables

Checking for missing values is as simple as running the query below. If your SQL developers didn't put NOT NULL constraints on your columns, this check will help. It's also useful when you're working with a lot of imported Excel files and want to ensure that all your customers have a phone number and email address.

SELECT COUNT(*) AS null_count

FROM orders

WHERE customer_id IS NULL

However, for a more automated approach, use the script below. Just pass in your table name and the script will find all the columns in the table. It will then loop through each one and give you a count of NULL values by column name. As an added bonus, you can pass a list of columns to exclude from the check.  Since some columns are expected to have NULLs, you can exclude them from the check.

from pyspark.sql import SparkSession

from pyspark.sql.functions import col

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_completeness(table_name, exclude_column=[]):

    df = spark.table(table_name)

    results = []

    for column in df.columns:

        if column in exclude_column:

            continue

        nulls = df.filter(col(column).isNull()).count()

        results.append((table_name, column, nulls))

    df = spark.createDataFrame(results, ["table", "column", "null_count"])

    return df

table_name = "finance"

df = test_completeness(table_name, ['primary_key', 'foreign_key'])

display(df)

2. Uniqueness: No Duplicate Keys

Here is another simple script to locate duplicate records for each column. Duplicate keys are by far the most frustrating problem in any data engineering project. One simple mistake can cause a many-to-many relationship and incorrect counts.  Sadly, such duplicates can be less than 0.1% of the table volume, making them tough to find using a spot check.

SELECT order_id

, COUNT(*) AS total

FROM orders

GROUP BY order_id

HAVING COUNT(*) > 1

Now let's improve this to let a user provide a table and column names and automatically check the result. You'll notice this script is very similar to the one in the first section, but instead of looking for NULL values, we're comparing distinct counts to total counts.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_uniqueness(table_name, columns):

    df = spark.table(table_name)

    total_count = df.count()

    results = []

    for column in columns:

        distinct_count = df.select(column).distinct().count()

        dupes = total_count - distinct_count

        results.append((table_name, column, dupes))

    df = spark.createDataFrame(results, ["table", "column", "duplicate_count"])

    return df

table_name = "finance"

df = test_uniqueness(table_name, ['foreign_key1', 'foreign_key2'])

display(df)

3. Referential Integrity: Ensure There Are No Orphaned Foreign Keys

This check is great for catching missing foreign keys. I've seen this issue often during testing. You refresh one table but forget to refresh a related table. This causes one table to end up with more keys than the other, or vice versa.

SELECT o.order_id

FROM orders o

LEFT OUTER JOIN customers c

ON o.customer_id = c.customer_id

WHERE c.customer_id IS NULL

The script below takes this a step further for a more automated process. Just supply two tables and the key that joins them. This script pulls all the keys into two lists, then compares them against each other to find keys that don't match.  The final results are returned as a DataFrame.

from pyspark.sql import SparkSession

from pyspark.sql.functions import lit

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_referential_integrity(left_table, right_table, left_key, right_key=None):

    left_df = spark.table(left_table)

    right_df = spark.table(right_table)

    # If no right_key is provided, assume it is the same as the left_key

    if right_key is None:

        right_key = left_key

    # Normalize keys to lists so single or composite keys both work

    left_key = [left_key] if isinstance(left_key, str) else left_key

    right_key = [right_key] if isinstance(right_key, str) else right_key

    join_condition = [

        left_df[lk] == right_df[rk] for lk, rk in zip(left_key, right_key)

    ]

    # Values in left_table with no match in right_table

    orphans = (

        left_df.join(right_df, on=join_condition, how="left_outer")

        .filter(right_df[right_key[0]].isNull())

        .select(left_df["*"])

        .withColumn("missing_from", lit(right_table)) )

    return orphans

left_table = "orders"

right_table = "customers"

left_key = "customer_id"

df = test_referential_integrity(left_table, right_table, left_key)

display(df)

4. Format Validity: Check for a Pattern Match

This check is extremely valuable when there's an expected pattern. For example, a phone number might be formatted (120) 234-3232 or 120.234.3232. An email address must include an "@" and a "." as shown in the example below. A Social Security number should be in the format 111-44-1221.

SELECT email

FROM customers

WHERE email NOT LIKE '%_@__%.__%'

Since spotting these discrepancies in a table of a million records is extremely difficult, here's a more automated approach.

from pyspark.sql import SparkSession

from pyspark.sql.functions import col

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_validity(table_name, column_name, pattern):

    df = spark.table(table_name)

    invalid_df = df.filter(~col(column_name).rlike(pattern)).select(column_name)

    return invalid_df

df = test_validity("customers", "email", r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$")

display(df)

Below is a table of common regex patterns you can use with this script. 

5. Data Freshness: Ensure Your Data Is Up to Date

This is the easiest but probably most useful check. If your data isn't current, your users will know and escalate quickly. Setting up checks to alert you early is critical. This gives you the opportunity to proactively warn users and stop the flood of angry emails in your inbox.

SELECT MAX(updated_at) AS last_update

FROM orders

Use the Python script below to find the maximum date across every table in a schema or database. Oftentimes, this is an updated_at or created_at column.  There is usually a standard naming convention for these columns in every database. 

from pyspark.sql import SparkSession

from pyspark.sql.functions import max as spark_max

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_freshness(schema_name, date_column="updated_at"):

    tables = spark.catalog.listTables(schema_name)

    results = []

    for t in tables:

        table_name = f"{schema_name}.{t.name}"

        df = spark.table(table_name)

        if date_column not in df.columns:

            results.append((table_name, None))

            continue

        last_update = df.agg(spark_max(date_column)).collect()[0][0]

        results.append((table_name, last_update))

    df = spark.createDataFrame(results, ["table", "last_update"])

    return df

df = test_freshness('finance', "updated_at")

display(df)

6. Schema Consistency: Ensure Data Matches Your Data Contract

The SQL query below gives you the data type of every column for a particular table. If you have a separate data contract, you can use this query to compare against it and easily locate discrepancies. Oftentimes, tables in the bronze layer ingest all columns as strings (or varchar) to expedite the process, but that approach can leave date fields stored as text instead of being converted to a proper date or integer datatype.

SELECT column_name

, data_type

FROM information_schema.columns

WHERE table_name = 'orders'

To level up your skill set, use the Python script below.

import pandas as pd

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_schema_consistency(schema_name, excel_path, sheet_name=0):

    # Load expected schema from Excel

    expected_df = pd.read_excel(excel_path, sheet_name=sheet_name)

    expected_df.columns = [c.strip().lower() for c in expected_df.columns]

    expected_lookup = {

        (row["table"], row["column"]): str(row["data_type"])

        for , row in expecteddf.iterrows()

    }

    # Pull actual schema for every table in the schema via the Spark catalog

    actual_lookup = {}

    for t in spark.catalog.listTables(schema_name):

        for c in spark.catalog.listColumns(t.name, schema_name):

            actual_lookup[(t.name, c.name)] = str(c.dataType)

    mismatches = []

    # Expected columns that are missing or have a mismatched data type

    for (table, column), expected_type in expected_lookup.items():

        actual_type = actual_lookup.get((table, column))

        if actual_type is None:

            mismatches.append((table, column, expected_type, None, "missing_column"))

        elif actual_type.lower() != expected_type.lower():

            mismatches.append((table, column, expected_type, actual_type, "type_mismatch"))

    # Columns that exist in the actual schema but aren't expected

    for (table, column), actual_type in actual_lookup.items():

        if (table, column) not in expected_lookup:

            mismatches.append((table, column, None, actual_type, "unexpected_column"))

    df = spark.createDataFrame(

        mismatches,

        ["table", "column", "expected_data_type", "actual_data_type", "issue"],

    )

    return df

df = test_schema_consistency('finance', 'finance_data_contract.xlsx')

display(df)

7. Anomaly Detection: Try to Find Outliers

This issue is much more challenging to find and more painful to troubleshoot. Sure, you can set constraints like in the SQL example below.

SELECT COUNT(*) AS error_count

FROM orders

WHERE order_date > GETUTCDATE()

OR order_date < '2000-01-01'

However, bad data can still have the correct data type. This is especially true of integer values. A value of 500 might be perfectly acceptable in one column, but that same value in another column could be a huge outlier. Think of this like sports statistics. A typical NFL stadium holds 60,000 to 80,000 people, most teams score around 20 to 30 points per game, and a quarterback may be sacked 1 to 3 times per game. All three of these values have very different expected ranges. A team scoring 14 points in a game is reasonable, but if the quarterback gets sacked 14 times, you may need to replace your offensive line.  The script below finds the median of every integer or float column.  It then calculates 3 standard deviations by default.  Any value outside of 3 standard deviations is considered an outlier. While these outliers may still be accurate values, narrowing the list down helps you find the truly outrageous records.  Furthermore, this example only works for a standard bell curve.  If your data skews to one side or the other, you will have to modify this script.

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, lit, percentile_approx, stddev

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

NUMERIC_TYPES = ("int", "bigint", "smallint", "tinyint", "float", "double", "decimal")

def test_outliers(table_name, std_dev_allowed=3):

    df = spark.table(table_name)

    numeric_columns = [

        field.name

        for field in df.schema.fields

        if any(field.dataType.simpleString().startswith(t) for t in NUMERIC_TYPES)

    ]

    results = []

    for column in numeric_columns:

        stats = df.select(

            percentile_approx(col(column), 0.5).alias("median"),

            stddev(col(column)).alias("std_dev"),

        ).collect()[0]

        median = stats["median"]

        std_dev = stats["std_dev"]

        # Skip columns that are all null or have zero variance

        if median is None or std_dev is None:

            continue

        three_std = std_dev_allowed * std_dev

        lower_bound = median - three_std

        upper_bound = median + three_std

        outliers = (

            df.filter((col(column) < lower_bound) | (col(column) > upper_bound))

            .select(col(column).cast("double").alias("value"))

            .withColumn("table_name", lit(table_name))

            .withColumn("column_name", lit(column))

            .withColumn("lower_limit", lit(float(lower_bound)))

            .withColumn("upper_limit", lit(float(upper_bound)))

            .select("table_name", "column_name", "value", "lower_limit", "upper_limit")

        )

        results.append(outliers)

    if not results:

        return spark.createDataFrame(

            [],

            "table_name string, column_name string, value double, "

            "lower_limit double, upper_limit double",

        )

    final_df = results[0]

    for r in results[1:]:

        final_df = final_df.unionByName(r)

    return final_df

8. Reconciliation: Check That Source and Target Tables Match

It's always wise to check whether your total row counts match between your source table and your target table. The SQL query below returns the row count for both tables.

SELECT

(SELECT COUNT(*) FROM source_db.orders) AS source_count

, (SELECT COUNT(*) FROM warehouse.orders) AS target_count

WHERE source_count <> target_count

The code below is more advanced. It hashes every shared field between both tables and joins on the key. This flags keys missing from either side, plus keys present in both tables with a mismatched value in another column. This row-by-row approach will find the most minor errors. You will often find issues if your data is not fresh. In this case, recent updates have not been applied to the target table, so they will be flagged.

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, sha2, concat_ws, when, lit

spark = SparkSession.builder.appName("SparkSession").getOrCreate()

def test_reconciliation(source_table, target_table, key_column):

    source_df = spark.table(source_table)

    target_df = spark.table(target_table)

    source_count = source_df.count()

    target_count = target_df.count()

    if source_count != target_count:

        print(f"Row count mismatch: {source_table}={source_count}, {target_table}={target_count}")

    # Only hash columns that exist on both sides, so a schema difference

    # doesn't silently throw off the comparison

    common_cols = sorted(set(source_df.columns) & set(target_df.columns))

    mismatched_cols = set(source_df.columns) ^ set(target_df.columns)

    if mismatched_cols:

        print(f"Columns not present on both sides, excluded from hash: {mismatched_cols}")

    def row_hash(df):

        # hash every shared column into a single value per row

        return sha2(concat_ws("|", *[col(c).cast("string") for c in common_cols]), 256)

    source_hashed = source_df.select(

        col(key_column).alias("key_value"),

        row_hash(source_df).alias("source_hash"),

    )

    target_hashed = target_df.select(

        col(key_column).alias("key_value"),

        row_hash(target_df).alias("target_hash"),

    )

    # Full outer join on the key so rows missing on either side show up

    # as NULL instead of just disappearing

    compared = source_hashed.join(target_hashed, on="key_value", how="full_outer")

    df = (

        compared.withColumn(

            "issue",

            when(col("source_hash").isNull(), lit("missing_from_source"))

            .when(col("target_hash").isNull(), lit("missing_from_target"))

            .when(col("source_hash") != col("target_hash"), lit("value_mismatch"))

            .otherwise(lit(None)),

        )

        .filter(col("issue").isNotNull())

        .select("key_value", "issue")

    )

    return df

df = test_reconciliation(source_table, target_table, key_column)

display(df)

Conclusion

None of these checks require a dedicated data quality platform to get started. Just open a notebook in Fabric or Databricks and copy these reusable functions. In my experience, I create a separate notebook to run at the end of a pipeline. Let the results write out to a log file. Then set up an automated email to the team when these errors occur.  I also create separate control tables to dynamically pass parameters into the functions.

If you would rather not build and maintain these checks yourself, there are established frameworks that do a lot of this for you. Great Expectations, DQX from Databricks Labs, DBT tests, Monte Carlo, and Soda Core all have built-in features to help. Depending on your existing tooling, any of these can replace or sit alongside the functions in this article.

These controls may take a little bit of time to set up, but the end gain is huge. You now own the process. No more frustrated users or emotional executives. Let your system work harder as you work smarter.