Data Quality Frameworks

by wshobson46891e7e60daNo licenseListed Oct 8, 2026Updated Oct 8, 2026

Implement data quality validation with Great Expectations, dbt tests, and data contracts. Use when building data quality pipelines, implementing validation rules, or establishing data contracts.

Instructions onlyData & Analytics

AI-generated overview

Guides implementing data quality validation with Great Expectations, dbt tests, and data contracts.

What it does: This skill provides production patterns for validating data pipelines using Great Expectations, dbt tests, and data contracts. It covers data quality dimensions such as completeness, uniqueness, validity, accuracy, consistency, and timeliness, along with a testing pyramid for data. It includes setup commands, expectation suite examples, a pipeline class that runs validations across tables and generates a pass/fail report, and best-practice guidance. Detailed pattern documentation is referenced in a separate file.
When to use it: Use it when adding data quality checks to pipelines, setting up Great Expectations validation, building dbt test suites, or establishing data contracts between teams. It also fits monitoring data quality metrics and automating validation in CI/CD.
Requirements: Requires Python and the great_expectations package, plus dbt for dbt tests; the skill is instructions only and ships no scripts. It references a separate details file for deeper patterns.

Data Quality Frameworks

Production patterns for implementing data quality with Great Expectations, dbt tests, and data contracts to ensure reliable data pipelines.

When to Use This Skill

Implementing data quality checks in pipelines
Setting up Great Expectations validation
Building comprehensive dbt test suites
Establishing data contracts between teams
Monitoring data quality metrics
Automating data validation in CI/CD

Core Concepts

1. Data Quality Dimensions

Dimension	Description	Example Check
Completeness	No missing values	`expect_column_values_to_not_be_null`
Uniqueness	No duplicates	`expect_column_values_to_be_unique`
Validity	Values in expected range	`expect_column_values_to_be_in_set`
Accuracy	Data matches reality	Cross-reference validation
Consistency	No contradictions	`expect_column_pair_values_A_to_be_greater_than_B`
Timeliness	Data is recent	`expect_column_max_to_be_between`

2. Testing Pyramid for Data

          /\         /  \     Integration Tests (cross-table)        /────\       /      \   Unit Tests (single column)      /────────\     /          \ Schema Tests (structure)    /────────────\

          /\         /  \     Integration Tests (cross-table)        /────\       /      \   Unit Tests (single column)      /────────\     /          \ Schema Tests (structure)    /────────────\

Quick Start

Great Expectations Setup

bash

# Installpip install great_expectations
# Initialize projectgreat_expectations init
# Create datasourcegreat_expectations datasource new

# Installpip install great_expectations
# Initialize projectgreat_expectations init
# Create datasourcegreat_expectations datasource new

python

# great_expectations/checkpoints/daily_validation.ymlimport great_expectations as gx
# Create contextcontext = gx.get_context()
# Create expectation suitesuite = context.add_expectation_suite("orders_suite")
# Add expectationssuite.add_expectation(    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))suite.add_expectation(    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
# Validateresults = context.run_checkpoint(checkpoint_name="daily_orders")

# great_expectations/checkpoints/daily_validation.ymlimport great_expectations as gx
# Create contextcontext = gx.get_context()
# Create expectation suitesuite = context.add_expectation_suite("orders_suite")
# Add expectationssuite.add_expectation(    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))suite.add_expectation(    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
# Validateresults = context.run_checkpoint(checkpoint_name="daily_orders")

Detailed patterns and worked examples

Detailed pattern documentation lives in references/details.md. Read that file when the navigation tier above is insufficient.

Summary: {total_passed}/{total_tables} tables passed")

    report.append("")
    for table, result in results.items():        status = "✅" if result.passed else "❌"        report.append(f"### {status} {table}")        report.append(f"- Expectations: {result.total_expectations}")        report.append(f"- Failed: {result.failed_expectations}")
        if not result.passed:            report.append("- Failed checks:")            for detail in result.details:                if not detail["success"]:                    report.append(f"  - {detail['expectation']}: {detail['observed_value']}")        report.append("")
    return "\n".join(report)

    report.append("")
    for table, result in results.items():        status = "✅" if result.passed else "❌"        report.append(f"### {status} {table}")        report.append(f"- Expectations: {result.total_expectations}")        report.append(f"- Failed: {result.failed_expectations}")
        if not result.passed:            report.append("- Failed checks:")            for detail in result.details:                if not detail["success"]:                    report.append(f"  - {detail['expectation']}: {detail['observed_value']}")        report.append("")
    return "\n".join(report)

Usage

context = gx.get_context() pipeline = DataQualityPipeline(context)

tables_to_validate = { "orders": "orders_suite", "customers": "customers_suite", "products": "products_suite", }

results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results)

Fail pipeline if any table failed

if not all(r.passed for r in results.values()): print(report) raise ValueError("Data quality checks failed!")


## Best Practices
### Do's
- **Test early** - Validate source data before transformations- **Test incrementally** - Add tests as you find issues- **Document expectations** - Clear descriptions for each test- **Alert on failures** - Integrate with monitoring- **Version contracts** - Track schema changes
### Don'ts
- **Don't test everything** - Focus on critical columns- **Don't ignore warnings** - They often precede failures- **Don't skip freshness** - Stale data is bad data- **Don't hardcode thresholds** - Use dynamic baselines- **Don't test in isolation** - Test relationships too


## Best Practices
### Do's
- **Test early** - Validate source data before transformations- **Test incrementally** - Add tests as you find issues- **Document expectations** - Clear descriptions for each test- **Alert on failures** - Integrate with monitoring- **Version contracts** - Track schema changes
### Don'ts
- **Don't test everything** - Focus on critical columns- **Don't ignore warnings** - They often precede failures- **Don't skip freshness** - Stale data is bad data- **Don't hardcode thresholds** - Use dynamic baselines- **Don't test in isolation** - Test relationships too

Source and attribution

Source:wshobson/agentsinplugins/data-engineering/skills/data-quality-frameworksat commit46891e7

License: No license

Content belongs to its original authors. SourceWeft indexes it from a public repository.

Report or request removal

Data Quality Frameworks

Data Quality Frameworks

When to Use This Skill

Core Concepts

1. Data Quality Dimensions

2. Testing Pyramid for Data

Quick Start

Great Expectations Setup

Detailed patterns and worked examples

Summary: {total_passed}/{total_tables} tables passed")

Usage

Fail pipeline if any table failed

Source and attribution

More from wshobson/agents

Uv Package Manager

Web Component Design

Python Type Safety

Visual Design Foundations

Responsive Design

Python Resource Management

More in Data & Analytics

Spanner Basics

Gke Cost Analysis

Datalineage Bigquery Asset Impact Analysis

Bigquery Troubleshooting

Bigquery Optimization

Bigquery Bigframes