ntb-dq-framework 0.1.5

Data Quality framework for Databricks — rule-based checks across six dimensions with Delta table persistence

NTB DQ Framework

Data Quality framework for Databricks — rule-based checks across six dimensions with Delta table persistence.

Installation

pip install ntb-dq-framework

Requires Python >= 3.9.

Quick Start

from pyspark.sql import SparkSession
from ntb_dq_framework import DQEngine

spark = SparkSession.builder.getOrCreate()

# Initialize — creates Delta tables if they don't exist yet
engine = DQEngine(spark, catalog="my_catalog")

# Register a rule
rule_id = engine.add_rule(
    rule_name="null_email",
    dimension="Completeness",
    target_table="my_catalog.my_schema.customers",
    target_column="email",
    rule_type="null_check",
    severity="critical",
    owner="data-team",
)

# Run all active rules for a table
summary = engine.run_checks("my_catalog.my_schema.customers")
print(f"Passed: {summary.total_rows_passed}, Failed: {summary.total_rows_failed}")

# Query results
results_df = engine.get_results(run_id=summary.run_id)
results_df.show()

Dimensions and Rule Types

Accuracy

Rule Type	Description	Key Parameters
reference_match	Joins target with a reference table and reports mismatched rows	reference_table, key_column, comparison_column
tolerance_match	Like reference_match but allows a numeric tolerance	reference_table, key_column, comparison_column, tolerance
reconciliation	Compares aggregate values (sum/count) between target and reference	reference_table, aggregate_column, aggregate_function

Completeness

Rule Type	Description	Key Parameters
null_check	Counts rows where the target column is null or empty	—
missing_partition	Compares expected date partitions against actual values	expected_partitions, partition_column
volume_check	Checks row count against a baseline threshold	baseline, threshold_percentage

Consistency

Rule Type	Description	Key Parameters
referential_integrity	Left anti-join to find orphan rows missing from a reference table	reference_table, key_column
cross_system_match	Joins on key columns and reports rows where compared columns differ	reference_table, key_columns, comparison_columns
business_rule	Evaluates a SQL boolean expression and reports rows where it's false	Uses rule_expression

Timeliness

Rule Type	Description	Key Parameters
freshness_check	Computes delay in hours since the latest timestamp value	—
sla_check	Fails when freshness delay exceeds a threshold	sla_threshold_hours

Validity

Rule Type	Description	Key Parameters
regex_check	Reports non-null rows not matching a regex pattern (nulls are excluded; column is auto-cast to string)	Uses rule_expression
allowed_values	Reports rows with values not in an allowed list	allowed_values
range_check	Reports rows outside min/max bounds	min, max
type_check	Reports rows that can't be cast to a target type	target_type

Uniqueness

Rule Type	Description	Key Parameters
duplicate_check	Groups by key columns and identifies duplicate groups	key_columns

API Reference

DQEngine

engine = DQEngine(spark, catalog="my_catalog", schema="my_schema")

Parameter	Type	Default	Description
spark	SparkSession	—	Active Spark session
catalog	str	—	Unity Catalog name
schema	str	"data_quality"	Schema for DQ tables. Optional — defaults to data_quality if omitted
table_prefix	str	"dq_"	Prefix for all Delta table names
max_failed_samples	int	100	Max failing rows to sample per rule

Method	Description
add_rule(...)	Register a rule. Returns the rule_id. Deduplicates identical configs automatically.
deactivate_rule(rule_id)	Soft-delete a rule by setting is_active = False.
run_checks(target_table, df=None, run_name=None, triggered_by="manual")	Execute all active rules for a table. Returns a RunSummary.
get_results(run_id=None, rule_id=None, dimension=None, target_table=None, date_range=None)	Query the results table with optional filters. Returns a Spark DataFrame.
daily_summary(run_date=None)	Aggregate run log data by date and target table.
dimension_summary(run_date=None)	Aggregate results by date, dimension, and target table.
rule_trend(rule_id=None, days=30)	Result data grouped by rule and date over a time window.
top_offenders(run_date=None, top_n=10)	Top N rules with the lowest pass rate.

Severity Levels

Rules support two severity levels: critical and warning.

• critical — raises DQRunError after all results are persisted if the rule fails or errors.
• warning — recorded in results but does not block execution.

Notes on rule_expression

When using regex patterns in rule_expression (e.g., for regex_check or business_rule), use a raw string to preserve backslashes:

engine.add_rule(
    rule_name="13_digit_id",
    dimension="Validity",
    target_table="my_catalog.my_schema.employees",
    target_column="id_card",
    rule_type="regex_check",
    rule_expression=r"^\d{13}$",  # raw string — backslashes are preserved
    severity="warning",
    owner="data-team",
)

Project Structure

ntb_dq_framework/
├── __init__.py            # Public API — exports DQEngine
├── engine.py              # DQEngine entry point
├── models.py              # RuleConfig, CheckResult, RunSummary dataclasses
├── rule_manager.py        # CRUD operations on the rule registry
├── run_executor.py        # Orchestrates check execution
├── result_writer.py       # Persists results to Delta tables
├── table_initializer.py   # Creates Delta tables on first run
├── monitoring.py          # Aggregation views for dashboards
├── exceptions.py          # DQValidationError
└── checks/
    ├── base.py            # BaseCheck abstract class
    ├── accuracy.py
    ├── completeness.py
    ├── consistency.py
    ├── timeliness.py
    ├── validity.py
    └── uniqueness.py

Delta Tables

On initialization, the framework creates four Delta tables (prefixed with dq_ by default):

Table	Purpose
dq_rule_registry	Stores rule definitions with activation status
dq_run_log	Tracks each run execution with summary stats
dq_results	Per-rule results for every run
dq_failed_records	Sampled failing rows (up to max_failed_samples)