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 | | ---------------- | ------------------------ | -------------------------------…

\"\n```\n\n### Pattern 4: Custom dbt Tests\n\n```sql\n-- tests/generic/test_row_count_in_range.sql\n{% test row_count_in_range(model, min_count, max_count) %}\n\nwith row_count as (\n select count(*) as cnt from {{ model }}\n)\n\nselect cnt\nfrom row_count\nwhere cnt \u003c {{ min_count }} or cnt > {{ max_count }}\n\n{% endtest %}\n\n-- Usage in schema.yml:\n-- tests:\n-- - row_count_in_range:\n-- min_count: 1000\n-- max_count: 10000000\n```\n\n```sql\n-- tests/generic/test_sequential_values.sql\n{% test sequential_values(model, column_name, interval=1) %}\n\nwith lagged as (\n select\n {{ column_name }},\n lag({{ column_name }}) over (order by {{ column_name }}) as prev_value\n from {{ model }}\n)\n\nselect *\nfrom lagged\nwhere {{ column_name }} - prev_value != {{ interval }}\n and prev_value is not null\n\n{% endtest %}\n```\n\n```sql\n-- tests/singular/assert_orders_customers_match.sql\n-- Singular test: specific business rule\n\nwith orders_customers as (\n select distinct customer_id from {{ ref('fct_orders') }}\n),\n\ndim_customers as (\n select customer_id from {{ ref('dim_customers') }}\n),\n\norphaned_orders as (\n select o.customer_id\n from orders_customers o\n left join dim_customers c using (customer_id)\n where c.customer_id is null\n)\n\nselect * from orphaned_orders\n-- Test passes if this returns 0 rows\n```\n\n### Pattern 5: Data Contracts\n\n```yaml\n# contracts/orders_contract.yaml\napiVersion: datacontract.com/v1.0.0\nkind: DataContract\nmetadata:\n name: orders\n version: 1.0.0\n owner: data-platform-team\n contact: [email protected]\n\ninfo:\n title: Orders Data Contract\n description: Contract for order event data from the ecommerce platform\n purpose: Analytics, reporting, and ML features\n\nservers:\n production:\n type: snowflake\n account: company.us-east-1\n database: ANALYTICS\n schema: CORE\n\nterms:\n usage: Internal analytics only\n limitations: PII must not be exposed in downstream marts\n billing: Charged per query TB scanned\n\nschema:\n type: object\n properties:\n order_id:\n type: string\n format: uuid\n description: Unique order identifier\n required: true\n unique: true\n pii: false\n\n customer_id:\n type: string\n format: uuid\n description: Customer identifier\n required: true\n pii: true\n piiClassification: indirect\n\n total_amount:\n type: number\n minimum: 0\n maximum: 100000\n description: Order total in USD\n\n created_at:\n type: string\n format: date-time\n description: Order creation timestamp\n required: true\n\n status:\n type: string\n enum: [pending, processing, shipped, delivered, cancelled]\n description: Current order status\n\nquality:\n type: SodaCL\n specification:\n checks for orders:\n - row_count > 0\n - missing_count(order_id) = 0\n - duplicate_count(order_id) = 0\n - invalid_count(status) = 0:\n valid values: [pending, processing, shipped, delivered, cancelled]\n - freshness(created_at) \u003c 24h\n\nsla:\n availability: 99.9%\n freshness: 1 hour\n latency: 5 minutes\n```\n\n### Pattern 6: Automated Quality Pipeline\n\n```python\n# quality_pipeline.py\nfrom dataclasses import dataclass\nfrom typing import List, Dict, Any\nimport great_expectations as gx\nfrom datetime import datetime\n\n@dataclass\nclass QualityResult:\n table: str\n passed: bool\n total_expectations: int\n failed_expectations: int\n details: List[Dict[str, Any]]\n timestamp: datetime\n\nclass DataQualityPipeline:\n \"\"\"Orchestrate data quality checks across tables\"\"\"\n\n def __init__(self, context: gx.DataContext):\n self.context = context\n self.results: List[QualityResult] = []\n\n def validate_table(self, table: str, suite: str) -> QualityResult:\n \"\"\"Validate a single table against expectation suite\"\"\"\n\n checkpoint_config = {\n \"name\": f\"{table}_validation\",\n \"config_version\": 1.0,\n \"class_name\": \"Checkpoint\",\n \"validations\": [{\n \"batch_request\": {\n \"datasource_name\": \"warehouse\",\n \"data_asset_name\": table,\n },\n \"expectation_suite_name\": suite,\n }],\n }\n\n result = self.context.run_checkpoint(**checkpoint_config)\n\n # Parse results\n validation_result = list(result.run_results.values())[0]\n results = validation_result.results\n\n failed = [r for r in results if not r.success]\n\n return QualityResult(\n table=table,\n passed=result.success,\n total_expectations=len(results),\n failed_expectations=len(failed),\n details=[{\n \"expectation\": r.expectation_config.expectation_type,\n \"success\": r.success,\n \"observed_value\": r.result.get(\"observed_value\"),\n } for r in results],\n timestamp=datetime.now()\n )\n\n def run_all(self, tables: Dict[str, str]) -> Dict[str, QualityResult]:\n \"\"\"Run validation for all tables\"\"\"\n results = {}\n\n for table, suite in tables.items():\n print(f\"Validating {table}...\")\n results[table] = self.validate_table(table, suite)\n\n return results\n\n def generate_report(self, results: Dict[str, QualityResult]) -> str:\n \"\"\"Generate quality report\"\"\"\n report = [\"# Data Quality Report\", f\"Generated: {datetime.now()}\", \"\"]\n\n total_passed = sum(1 for r in results.values() if r.passed)\n total_tables = len(results)\n\n report.append(f\"\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":11648,"content_sha256":"224757dd2c96e2c13fddea88cfe74297f35bf7aae00ec1622bfec0585210f812"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Data Quality Frameworks","type":"text"}]},{"type":"paragraph","content":[{"text":"Production patterns for implementing data quality with Great Expectations, dbt tests, and data contracts to ensure reliable data pipelines.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"When to Use This Skill","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Implementing data quality checks in pipelines","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Setting up Great Expectations validation","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Building comprehensive dbt test suites","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Establishing data contracts between teams","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Monitoring data quality metrics","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Automating data validation in CI/CD","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Core Concepts","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"1. Data Quality Dimensions","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Dimension","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Description","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Example Check","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Completeness","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"No missing values","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"expect_column_values_to_not_be_null","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Uniqueness","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"No duplicates","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"expect_column_values_to_be_unique","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Validity","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Values in expected range","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"expect_column_values_to_be_in_set","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Accuracy","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Data matches reality","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Cross-reference validation","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Consistency","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"No contradictions","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"expect_column_pair_values_A_to_be_greater_than_B","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Timeliness","type":"text","marks":[{"type":"strong"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Data is recent","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"expect_column_max_to_be_between","type":"text","marks":[{"type":"code_inline"}]}]}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"2. Testing Pyramid for Data","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":" /\\\n / \\ Integration Tests (cross-table)\n /────\\\n / \\ Unit Tests (single column)\n /────────\\\n / \\ Schema Tests (structure)\n /────────────\\","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Quick Start","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Great Expectations Setup","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Install\npip install great_expectations\n\n# Initialize project\ngreat_expectations init\n\n# Create datasource\ngreat_expectations datasource new","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# great_expectations/checkpoints/daily_validation.yml\nimport great_expectations as gx\n\n# Create context\ncontext = gx.get_context()\n\n# Create expectation suite\nsuite = context.add_expectation_suite(\"orders_suite\")\n\n# Add expectations\nsuite.add_expectation(\n gx.expectations.ExpectColumnValuesToNotBeNull(column=\"order_id\")\n)\nsuite.add_expectation(\n gx.expectations.ExpectColumnValuesToBeUnique(column=\"order_id\")\n)\n\n# Validate\nresults = context.run_checkpoint(checkpoint_name=\"daily_orders\")","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Detailed patterns and worked examples","type":"text"}]},{"type":"paragraph","content":[{"text":"Detailed pattern documentation lives in ","type":"text"},{"text":"references/details.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". Read that file when the navigation tier above is insufficient.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Summary: {total_passed}/{total_tables} tables passed\")","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":" report.append(\"\")\n\n for table, result in results.items():\n status = \"✅\" if result.passed else \"❌\"\n report.append(f\"### {status} {table}\")\n report.append(f\"- Expectations: {result.total_expectations}\")\n report.append(f\"- Failed: {result.failed_expectations}\")\n\n if not result.passed:\n report.append(\"- Failed checks:\")\n for detail in result.details:\n if not detail[\"success\"]:\n report.append(f\" - {detail['expectation']}: {detail['observed_value']}\")\n report.append(\"\")\n\n return \"\\n\".join(report)","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Usage","type":"text"}]},{"type":"paragraph","content":[{"text":"context = gx.get_context() pipeline = DataQualityPipeline(context)","type":"text"}]},{"type":"paragraph","content":[{"text":"tables_to_validate = { \"orders\": \"orders_suite\", \"customers\": \"customers_suite\", \"products\": \"products_suite\", }","type":"text"}]},{"type":"paragraph","content":[{"text":"results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results)","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Fail pipeline if any table failed","type":"text"}]},{"type":"paragraph","content":[{"text":"if not all(r.passed for r in results.values()): print(report) raise ValueError(\"Data quality checks failed!\")","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"\n## Best Practices\n\n### Do's\n\n- **Test early** - Validate source data before transformations\n- **Test incrementally** - Add tests as you find issues\n- **Document expectations** - Clear descriptions for each test\n- **Alert on failures** - Integrate with monitoring\n- **Version contracts** - Track schema changes\n\n### Don'ts\n\n- **Don't test everything** - Focus on critical columns\n- **Don't ignore warnings** - They often precede failures\n- **Don't skip freshness** - Stale data is bad data\n- **Don't hardcode thresholds** - Use dynamic baselines\n- **Don't test in isolation** - Test relationships too","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"data-quality-frameworks","author":"@skillopedia","source":{"stars":42106,"repo_name":"agents","origin_url":"https://github.com/wshobson/agents/blob/HEAD/plugins/data-engineering/skills/data-quality-frameworks/SKILL.md","repo_owner":"wshobson","body_sha256":"58e61444165ba7caa572fb7f1ba2d23cfd5038cfa6699f86b5648e7fcb71d34d","cluster_key":"d898c537e2d6b6b403bfb25ea01837b9e7f3c29d1397f8ced711993ed3a8c15f","clean_bundle":{"format":"clean-skill-bundle-v1","source":"wshobson/agents/plugins/data-engineering/skills/data-quality-frameworks/SKILL.md","text_bytes":11648,"source_commit":"7e2fb559cf4b824b27f1fec49fe88f7511066d26","skills_sh_rank":2157,"attachment_count":1,"text_attachments":1,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"plugins/data-engineering/skills/data-quality-frameworks/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"data-analytics","category_label":"Data"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"data-analytics","import_tag":"skills-sh-top20k-2026-11","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."}},"renderedAt":1793665970711}

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 | | ---------------- | ------------------------ | -------------------------------…