| name | data-quality-frameworks |
| description | 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. |
| version | 1.0.0 |
| cluster | databases-data |
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
pip install great_expectations
great_expectations init
great_expectations datasource new
import great_expectations as gx
context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
results = context.run_checkpoint(checkpoint_name="daily_orders")
Patterns
Pattern 1: Great Expectations Suite
import great_expectations as gx
from great_expectations.core import ExpectationSuite
from great_expectations.core.expectation_configuration import ExpectationConfiguration
def build_orders_suite() -> ExpectationSuite:
"""Build comprehensive orders expectation suite"""
suite = ExpectationSuite(expectation_suite_name="orders_suite")
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_table_columns_to_match_set",
kwargs={
"column_set": ["order_id", "customer_id", "amount", "status", "created_at"],
"exact_match": False
}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_not_be_null",
kwargs={"column": "order_id"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_be_unique",
kwargs={"column": "order_id"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_not_be_null",
kwargs={"column": "customer_id"}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type="expect_column_values_to_be_in_set",
kwargs={
"column": "status",
: [, , , , ]
}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type=,
kwargs={
: ,
: ,
: ,
:
}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type=,
kwargs={: }
))
suite.add_expectation(ExpectationConfiguration(
expectation_type=,
kwargs={
: ,
: {: },
: {: }
}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type=,
kwargs={
: ,
:
}
))
suite.add_expectation(ExpectationConfiguration(
expectation_type=,
kwargs={
: ,
: ,
:
}
))
suite
Pattern 2: Great Expectations Checkpoint
name: orders_checkpoint
config_version: 1.0
class_name: Checkpoint
run_name_template: "%Y%m%d-%H%M%S-orders-validation"
validations:
- batch_request:
datasource_name: warehouse
data_connector_name: default_inferred_data_connector_name
data_asset_name: orders
data_connector_query:
index: -1
expectation_suite_name: orders_suite
action_list:
- name: store_validation_result
action:
class_name: StoreValidationResultAction
- name: store_evaluation_parameters
action:
class_name: StoreEvaluationParametersAction
- name: update_data_docs
action:
class_name: UpdateDataDocsAction
- name: send_slack_notification
action:
class_name: SlackNotificationAction
import great_expectations as gx
context = gx.get_context()
result = context.run_checkpoint(checkpoint_name="orders_checkpoint")
if not result.success:
failed_expectations = [
r for r in result.run_results.values()
if not r.success
]
raise ValueError(f"Data quality check failed: {failed_expectations}")
Pattern 3: dbt Data Tests
version: 2
models:
- name: fct_orders
description: Order fact table
tests:
- dbt_utils.recency:
datepart: day
field: created_at
interval: 1
- dbt_utils.at_least_one
- dbt_utils.expression_is_true:
expression: "total_amount >= 0"
columns:
- name: order_id
description: Primary key
tests:
- unique
- not_null
- name: customer_id
description: Foreign key to dim_customers
tests:
- not_null
- relationships:
to: ref('dim_customers')
[, , , , ]
Pattern 4: Custom dbt Tests
{% test row_count_in_range(model, min_count, max_count) %}
with row_count as (
select count(*) as cnt from {{ model }}
)
select cnt
from row_count
where cnt < {{ min_count }} or cnt > {{ max_count }}
{% endtest %}
{% test sequential_values(model, column_name, interval=1) %}
with lagged as (
select
{{ column_name }},
lag({{ column_name }}) over (order by {{ column_name }}) as prev_value
from {{ model }}
)
select *
from lagged
where {{ column_name }} - prev_value != {{ interval }}
and prev_value is not null
{% endtest %}
with orders_customers as (
select distinct customer_id from {{ ref('fct_orders') }}
),
dim_customers as (
select customer_id from {{ ref('dim_customers') }}
),
orphaned_orders as (
select o.customer_id
from orders_customers o
left join dim_customers c using (customer_id)
where c.customer_id is null
)
select * from orphaned_orders
Pattern 5: Data Contracts
apiVersion: datacontract.com/v1.0.0
kind: DataContract
metadata:
name: orders
version: 1.0.0
owner: data-platform-team
contact: data-team@company.com
info:
title: Orders Data Contract
description: Contract for order event data from the ecommerce platform
purpose: Analytics, reporting, and ML features
servers:
production:
type: snowflake
account: company.us-east-1
database: ANALYTICS
schema: CORE
terms:
usage: Internal analytics only
limitations: PII must not be exposed
[, , , , ]
[, , , , ]
Pattern 6: Automated Quality Pipeline
from dataclasses import dataclass
from typing import List, Dict, Any
import great_expectations as gx
from datetime import datetime
@dataclass
class QualityResult:
table: str
passed: bool
total_expectations: int
failed_expectations: int
details: List[Dict[str, Any]]
timestamp: datetime
class DataQualityPipeline:
"""Orchestrate data quality checks across tables"""
def __init__(self, context: gx.DataContext):
self.context = context
self.results: List[QualityResult] = []
def validate_table(self, table: str, suite: str) -> QualityResult:
"""Validate a single table against expectation suite"""
checkpoint_config = {
"name": f"{table}_validation",
"config_version": 1.0,
"class_name": "Checkpoint",
"validations": [{
"batch_request": {
: ,
: table,
},
: suite,
}],
}
result = .context.run_checkpoint(**checkpoint_config)
validation_result = (result.run_results.values())[]
results = validation_result.results
failed = [r r results r.success]
QualityResult(
table=table,
passed=result.success,
total_expectations=(results),
failed_expectations=(failed),
details=[{
: r.expectation_config.expectation_type,
: r.success,
: r.result.get(),
} r results],
timestamp=datetime.now()
)
() -> [, QualityResult]:
results = {}
table, suite tables.items():
()
results[table] = .validate_table(table, suite)
results
() -> :
report = [, , ]
total_passed = ( r results.values() r.passed)
total_tables = (results)
report.append()
report.append()
table, result results.items():
status = result.passed
report.append()
report.append()
report.append()
result.passed:
report.append()
detail result.details:
detail[]:
report.append()
report.append()
.join(report)
context = gx.get_context()
pipeline = DataQualityPipeline(context)
tables_to_validate = {
: ,
: ,
: ,
}
results = pipeline.run_all(tables_to_validate)
report = pipeline.generate_report(results)
(r.passed r results.values()):
(report)
ValueError()
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
Resources