ntb-dq-framework 1.1.0

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.

Claude Code Agent (optional)

The package ships with a Claude Code agent that helps you design rules, write correct add_rule() calls, and debug DQ results. After installing the library, run:

ntb-dq install-agent              # project-level: ./.claude/agents/
ntb-dq install-agent --user       # user-level (all projects): ~/.claude/agents/
ntb-dq install-agent --force      # overwrite an existing agent file

The agent activates automatically in Claude Code when you mention DQEngine, ntb_dq_framework, or any of the six dimensions. No copy-paste required — ntb-dq install-agent reads the bundled agent file from the installed package (works even when the venv is gitignored).

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

How rules work: You always define what valid (good) data looks like. The framework then reports any rows that don't match as failures. For example, allowed_values=["Male", "Female"] means rows with those values pass — anything else fails. This applies to all rule types: regex patterns describe valid formats, range bounds describe acceptable values, business rules describe conditions that should be true, etc.

Accuracy

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

source_* params default to their target_* counterparts when omitted, so same-name keys (e.g. orders.customer_id ↔ customer_dim.customer_id) only need the target_* side.

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, target_key_column, source_key_column
cross_system_match	Joins on key columns and reports rows where compared columns differ	reference_table, target_key_columns, source_key_columns, target_comparison_columns, source_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. See add_rule Parameters below.
deactivate_rule(rule_id)	Soft-delete a rule by setting is_active = False.
run_checks(target_table, df=None)	Execute all active rules for a table. Returns a RunSummary (includes critical_failed_rows_df — see Filtering Critical Failures).
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.

add_rule Parameters

engine.add_rule(
    rule_name="...",
    dimension="...",
    target_table="...",
    rule_type="...",
    rule_expression="",        # SQL expression or regex pattern (for regex_check, business_rule)
    target_column=None,        # Column to check
    parameters=None,           # Dict of rule-type-specific parameters (see below)
    severity="warning",        # "critical" or "warning"
    owner="",
)

Parameter	Type	Default	Description
rule_name	str	—	Human-readable name for the rule
dimension	str	—	One of Accuracy, Completeness, Consistency, Timeliness, Validity, Uniqueness
target_table	str	—	Fully qualified table name (e.g., catalog.schema.table)
rule_type	str	—	Check type (see Dimensions and Rule Types)
rule_expression	str	""	SQL expression or regex pattern (used by regex_check, business_rule)
target_column	str | None	None	Column to evaluate
parameters	dict | None	None	Rule-type-specific parameters (see Key Parameters in each dimension table)
severity	str	"warning"	"critical" or "warning"
owner	str	""	Owner or team responsible for this rule

Note: Rule-type-specific options like allowed_values, min, max, key_columns, etc. must be passed inside the parameters dict — not as top-level keyword arguments.

Examples:

# Validity — allowed_values
engine.add_rule(
    rule_name="valid_gender",
    dimension="Validity",
    target_table="my_catalog.my_schema.customers",
    target_column="gender",
    rule_type="allowed_values",
    severity="warning",
    owner="data-team",
    parameters={"allowed_values": ["Male", "Female"]},
)

# Validity — range_check
engine.add_rule(
    rule_name="valid_age",
    dimension="Validity",
    target_table="my_catalog.my_schema.customers",
    target_column="age",
    rule_type="range_check",
    parameters={"min": 0, "max": 120},
)

# Uniqueness — duplicate_check
engine.add_rule(
    rule_name="unique_order",
    dimension="Uniqueness",
    target_table="my_catalog.my_schema.orders",
    rule_type="duplicate_check",
    parameters={"key_columns": ["order_id"]},
)

Severity Levels

Rules support two severity levels: critical and warning.

• critical — all failed rows are saved to dq_failed_records (no sampling cap) and collected into RunSummary.critical_failed_rows_df so callers can filter them out before writing.
• warning — only a sample of failed rows (up to max_failed_samples) is saved to dq_failed_records.

Filtering Critical Failures

run_checks() returns a RunSummary with critical_failed_rows_df — the full (uncapped) union of all failed rows from critical-severity rules. Use it to filter out bad records before upserting:

summary = engine.run_checks("my_catalog.my_schema.customers")

if summary.critical_failed_rows_df is not None:
    clean_df = df.join(summary.critical_failed_rows_df, how="left_anti")
    clean_df.write.mode("overwrite").saveAsTable("my_catalog.my_schema.customers")
else:
    df.write.mode("overwrite").saveAsTable("my_catalog.my_schema.customers")

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",
)

Scoping a rule with filter_expression

Any rule can be restricted to a subset of rows by adding a filter_expression to its parameters. It is a SQL boolean expression: only rows where it evaluates TRUE are checked. Rows excluded by the filter are counted as passing — the run keeps the full-table row count as the denominator, so the filter never shrinks rows_evaluated.

This works for every rule type (including null_check, duplicate_check, range_check, etc. that have no expression of their own), so it is the way to skip soft-deleted rows without pre-filtering the DataFrame yourself:

# Only check live rows; soft-deleted rows are treated as passing.
engine.add_rule(
    rule_name="email_not_null",
    dimension="Completeness",
    target_table="my_catalog.my_schema.customers",
    target_column="email",
    rule_type="null_check",
    severity="critical",
    parameters={"filter_expression": "COALESCE(is_deleted, false) = false"},
)

Notes:

• Wrap a nullable flag in COALESCE(...) so rows with a NULL flag aren't mis-scoped.
• The expression goes through the same SQL guard as rule_expression; an unsafe expression raises DQValidationError at add_rule time.
• It applies to the target DataFrame only (for reference-based checks like referential_integrity / cross_system_match, the reference table is not filtered).
• For a filter that should apply to all rules of a run, pre-filter the DataFrame and pass it via run_checks(target_table, df=...) instead.

SQL Guard — Blocked Patterns

The framework includes a built-in SQL guard that validates all user-supplied expressions and identifiers before they reach Spark SQL. If a blocked pattern is detected, a DQValidationError is raised.

Blocked DDL/DML statements:

DROP, ALTER, CREATE, INSERT, UPDATE, DELETE, TRUNCATE, MERGE, GRANT, REVOKE

Blocked injection patterns:

Pattern	Example
SQL comments	--, /* */
UNION SELECT	UNION ALL SELECT ...
INTO OUTFILE/DUMPFILE	File exfiltration
EXEC() / EXECUTE()	Stored procedure execution
xp_* / sp_*	SQL Server system procedures

Allowed — normal SQL boolean expressions:

"total_amount > 0"
"end_date >= start_date"
"status IN ('active', 'pending')"
"price * quantity = total AND discount >= 0"
"COALESCE(email, '') != ''"

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 (partitioned by target_table)
dq_result_table	Per-rule results for every run (partitioned by target_table)
dq_failed_records	Failing rows — all rows for critical rules, sampled up to max_failed_samples for warning rules (partitioned by target_table)

Changelog

0.4.0

• Bundled a Claude Code agent (ntb-dq-framework.md) inside the package and added an ntb-dq console script. Run ntb-dq install-agent (project) or ntb-dq install-agent --user (global) after pip install to drop the agent into .claude/agents/. The agent activates automatically in Claude Code when you mention DQEngine or any of the six dimensions, and works even when the venv is gitignored.

0.3.1

• Breaking: primary-key columns on the four dq_* tables are now STRING UUID4 values, not BIGINT IDENTITY. add_rule() returns a str; deactivate_rule(rule_id), get_results(run_id=, rule_id=), and rule_trend(rule_id=) all take strings.
• Migration: drop the existing tables once before deploying — DROP TABLE catalog.data_quality.dq_rule_registry, dq_run_log, dq_result_table, dq_failed_records. They are recreated with the new schema on the next DQEngine initialization. Rule definitions must be re-registered.
• Why: removes the MetadataChangedException: ... "delta.identity.schemaUpdate":"true" raised by parallel run_checks invocations. Identity-column allocation mutates table metadata on every append, so concurrent writers conflicted even with WriteSerializable isolation and target_table partitioning (added in 0.2.1). UUIDs are generated in Python at insert time, so writes are pure data appends with no metadata mutation. The racy SELECT id ORDER BY id DESC LIMIT 1 readbacks in ResultWriter and RuleManager are also gone.

0.2.2

• Fixed DQRunError: [INCOMPATIBLE_COLUMN_TYPE] UNION ... when combining a duplicate_check with other row-level checks. The run-level aggregation of failed DataFrames now uses unionByName(allowMissingColumns=True) instead of positional unionAll, tolerating column reordering (e.g. from the duplicate-check join) and minor schema differences across rules.

0.2.1

• run_checks now returns critical_failed_rows_df on RunSummary — the full (uncapped) union of all failed rows from critical-severity rules, so callers can filter bad records before upserting.
• Critical-severity rules now save all failed rows to dq_failed_records (no sampling cap). Warning rules still save up to max_failed_samples.
• Delta tables (dq_run_log, dq_result_table, dq_failed_records) are now partitioned by target_table with WriteSerializable isolation level, enabling parallel pipeline execution without MetadataChangedException.

0.2.0

• run_checks no longer raises DQRunError when critical-severity rules fail or error. The method now returns a RunSummary containing all results (including critical failures), allowing the pipeline to continue. Results are still persisted to Delta tables as before.

0.1.5

• Initial release with six-dimension rule-based checks and Delta table persistence.