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.

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 Quick Start Great Expectations Setup 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

data-quality-frameworks