Data Quality Frameworks

When to use this skill

Use Data Quality Frameworks when an AI agent needs a reusable SKILL.md workflow for this job: 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.

When not to use it

Skip Data Quality Frameworks when the task is outside the coding category, or when a more specific skill in this directory already covers the same workflow with clearer triggers.

How to install

1. Personal install: create ~/.claude/skills/data-quality-frameworks/SKILL.md (and any bundled scripts) so Claude Code, Claude Desktop, and compatible agents can load it in every project.
2. Project install: commit the same folder at .claude/skills/data-quality-frameworks/ so teammates get the skill with the repo.
3. Restart the agent session after copying files so it re-scans the skills directory, then ask for the task in words that match the skill description.

Full install guide for Claude, Cursor, and Codex

What this skill does

# 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) /────────────\ ```

## Quick Start

### Great Expectations Setup

```bash # Install pip install great_expectations

# Initialize project great_expectations init

# Create datasource great_expectations datasource new ```

```python # great_expectations/checkpoints/daily_validation.yml import great_expectations as gx

# Create context context = gx.get_context()

# Create expectation suite suite = context.add_expectation_suite("orders_suite")

# Add expectations suite.add_expectation( gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id") ) suite.add_expectation( gx.expectations.ExpectColumnValuesToBeUnique(column="order_id") )

# Validate results = 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)

# 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

Intended uses

• 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