j4flmao Data Data Quality
Purpose
Build a data quality framework covering quality dimensions (completeness, accuracy, timeliness, consistency, uniqueness, integrity), automated validation tests (Great Expectations expectations suites, data docs, checkpoints; dbt singular, generic, freshness tests), data observability (Soda, Monte Carlo, Elementary), data SLAs with escalation paths, and data contracts.
Agent Protocol
Trigger
Exact user phrases: "data quality", "data validation", "data profiling", "Great Expectations", "dbt tests", "data observability", "data contract", "schema validation", "data quality check", "data testing", "data monitoring", "quality dimensions", "data freshness", "data completeness", "Soda", "Monte Carlo", "data SLA", "data integrity".
Input Context
Before activating, verify:
- Data stack (warehouse, lake, streaming platform)
- Transformation tool (dbt, Spark, custom SQL)
- Data sources and producers (internal, external, partner)
- Existing monitoring and alerting infrastructure
- Critical data assets for business operations
- Data consumers and their quality SLAs
Output Artifact
Data quality framework with dimension definitions, test configurations, monitoring setup, and contract templates.
Response Format
# Quality dimension definitions
# Great Expectations suite
# dbt test config
# Data contract template
# Alert rules
# SLA definitions
No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.
Completion Criteria
- Quality dimensions defined with measurement approach
- Automated validation suite (Great Expectations + dbt)
- Data profiling established for baseline expectations
- Data observability with monitoring and alerting
- Data contracts between producers and consumers
- Quality SLAs documented with escalation paths
- Soda checks configured for in-storage validation
- Data integrity checks across referential relationships
Max Response Length
300 lines of configuration.
Workflow
Step 1: Quality Dimensions
Completeness: percentage of non-null values for required columns.
Measure: COUNT(column) / COUNT(*) * 100.
Critical columns: 100% required.
High: over 99% for business-required fields.
Accuracy: values represent real-world facts. Cross-reference order total against sum of line items. Range validation: age between 0 and 150, price above 0.
Timeliness: data freshness against expected SLA.
Measure: CURRENT_TIMESTAMP - MAX(updated_at).
Critical: within 5 minutes. High: within 1 hour.
Consistency: same values across different systems. Compare counts between source and warehouse. Check date formats and currency conversions.
Uniqueness: no duplicate records for defined keys.
Measure: COUNT(*) - COUNT(DISTINCT key).
Primary keys: 0% duplicate rate.
Integrity: referential integrity checks. Every foreign key exists in parent table. No orphan records or dangling references.
Validity: conforms to format and domain rules. Email regex, ISO date format, enum values.
Weight each dimension for composite score. Default: completeness 20%, accuracy 25%, consistency 15%, timeliness 15%, uniqueness 10%, validity 15%.
Step 2: Great Expectations
Expectation suites per critical table. Types: not_null, unique, between (range), match_regex (format), pair_equal (cross-column).
Profiling: auto-generate from sample data. Analyzes null rate, min/max, distinct values, value frequency, type inference. Review and adjust before production deployment.
Validation: run as checkpoint in CI pipeline.
great_expectations checkpoint run orders_checkpoint.
Block pipeline on critical expectation failures.
Data docs: auto-generated HTML documentation. Expectation results per run, validation timeline, data source overview. Host on S3 or GCS.
Rules: every critical table has 10+ expectations.
Step 2a: Great Expectations Suite YAML
# expectations/orders_suite.yml
expectations:
- expectation_type: expect_column_values_to_not_be_null
kwargs: { column: order_id, mostly: 1.0 }
meta: { dimension: completeness, critical: true }
- expectation_type: expect_column_values_to_be_unique
kwargs: { column: order_id }
meta: { dimension: uniqueness, critical: true }
- expectation_type: expect_column_values_to_be_between
kwargs: { column: total_amount, min_value: 0, max_value: 100000 }
meta: { dimension: accuracy, critical: true }
- expectation_type: expect_table_row_count_to_be_between
kwargs: { min_value: 1000, max_value: 5000000 }
meta: { dimension: volume }
- expectation_type: expect_column_pair_values_to_be_equal
kwargs: { column_A: currency, column_B: currency_code }
meta: { dimension: consistency }
# checkpoints/orders_checkpoint.yml
name: orders_checkpoint
config_version: 1.0
class_name: SimpleCheckpoint
validations:
- batch_request:
datasource_name: warehouse
data_asset_name: analytics.fct_orders
expectation_suite_name: orders_suite
action_list:
- name: store_validation_result
action: { class_name: StoreValidationResultAction }
- name: update_data_docs
action: { class_name: UpdateDataDocsAction }
- name: send_slack_notification
action:
class_name: SlackNotificationAction
slack_webhook: ${SLACK_WEBHOOK}
notify_on: failure
Step 3: dbt Tests
Generic tests: unique, not_null, accepted_values, relationships (referential integrity).
Custom generic tests:
- freshness: age of latest record versus threshold
- row_count_threshold: minimum and maximum rows
- distribution_outlier: value distribution drift
Singular tests: custom SQL returning failing rows.
SELECT * FROM fct_orders WHERE total_amount < 0.
CI integration:
dbt test --select tag:critical blocks pipeline.
dbt test --select severity:warn informational only.
dbt test --store_failures for trend analysis.
Coverage: every column in critical tables has one test.
Step 3a: dbt Test Definitions YAML
# models/schema.yml — generic and singular tests
version: 2
models:
- name: fct_orders
tests:
- dbt_utils.unique_combination_of_columns:
combination_of_columns: [order_id, line_item_id]
columns:
- name: order_id
tests:
- unique
- not_null
- relationships:
to: ref('dim_orders')
field: order_id
- name: total_amount
tests:
- not_null
- dbt_utils.accepted_range:
min_value: 0
max_value: 100000
where: "status = 'completed'"
- name: status
tests:
- accepted_values:
values: [pending, completed, cancelled, refunded]
- name: created_at
tests:
- fresh:
warn_after: { count: 12, period: hour }
error_after: { count: 24, period: hour }
-- singular test: negative amounts
SELECT order_id, total_amount
FROM {{ ref('fct_orders') }}
WHERE total_amount < 0;
Run: dbt test --select tag:critical --store_failures
Step 4: Soda Checks
In-storage quality checks with SQL-based metrics. YAML configuration per table.
Check types: freshness under N hours, row count minimum, missing percentage threshold, duplicate percentage equals 0, schema change detection.
Execution: soda scan -d warehouse -c config.yml checks.yml.
Soda Cloud for visualization and alerting.
Open-source CLI for CI pipeline integration.
Rules: every production table has checks before deployment.
Step 4a: Soda Check YAML Examples
# checks/orders_checks.yml
checks for analytics.fct_orders:
- freshness(created_at) < 6h
- row_count between 10000 and 5000000
- duplicate_count(order_id) = 0
- schema:
name: Schema validation
fail:
when required column missing: [order_id, customer_id, total_amount]
when wrong column type:
order_id: string
total_amount: numeric
- missing_percent(customer_id) < 0.1
- avg(total_amount) > 0
- min(total_amount) >= 0
- max(total_amount) <= 100000
- values in (status) must_exist_in [pending, completed, cancelled, refunded]
checks for analytics.fct_orders:
- values in (customer_id) must_exist_in analytics.dim_customers (customer_id)
- anomaly detection for row_count:
sensitivity: normal
training_period: 30
# configuration.yml
data_source warehouse:
type: postgres
connection:
host: ${POSTGRES_HOST}
port: 5432
database: analytics
schema: analytics
soda_cloud:
host: cloud.soda.io
api_key_id: ${SODA_API_KEY_ID}
api_key_secret: ${SODA_API_KEY_SECRET}
Run: soda scan -d warehouse -c configuration.yml checks/orders_checks.yml
Step 5: Data Contracts
Contract between producer and consumer.
Fields: table schema (columns, types, constraints), freshness SLA (updated by 6am daily), row count range (min 1000, max 10M), nullability thresholds (over 99.5% complete), integrity rules (referential checks).
Producer signs contract. Consumer validates at ingestion.
Enforcement chain: Schema validation at ingestion. Great Expectations at landing. dbt tests at transformation. Soda freshness monitors continuous.
Breach action: Alert producer and consumer via Slack or PagerDuty. Pause downstream data pipelines. Document in quality dashboard.
Contract versioned alongside pipeline code. Changes require both parties to approve.
Step 5a: Automated Quality Gate CI Config
# .github/workflows/quality-gate.yml
jobs:
quality-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Great Expectations Validation
run: great_expectations checkpoint run orders_checkpoint
- name: dbt Test Suite
run: dbt test --select tag:critical --store_failures
- name: Soda Scan
run: soda scan -d warehouse -c configuration.yml checks/orders_checks.yml
- name: Upload Quality Report
if: always()
run: |
aws s3 sync great_expectations/uncommitted/data_docs/ s3://ge-data-docs/
- name: Block Pipeline on Failure
if: failure()
run: |
curl -X POST -H "Content-type: application/json" \
--data '{"text":"Quality gate failed for fct_orders — pipeline blocked"}' \
${{ secrets.SLACK_WEBHOOK }}
Gate blocks deploy on: critical GE expectation failure, dbt test failure on tag:critical, Soda scan finding invalid schema or referential integrity violation. Warnings pass through but log to quality dashboard.
Step 6: Observability (Monte Carlo / Elementary)
Key metrics: Freshness: time since last successful load. Volume: row count versus expected range. Schema: new, missing, or renamed columns. Quality: test pass rate over time. Lineage: data flow from source to consumption.
Monte Carlo: SaaS end-to-end observability. ML-based anomaly detection, no manual config.
Elementary: open-source, dbt-native. Quality dashboard and Slack alerts. Column-level lineage and impact analysis.
Dashboard: quality score per table, pass rate trends, oldest failing test, schema change history. Alerts: Slack for warnings, PagerDuty for critical.
Step 7: Data SLAs and Escalation
Tiers: Critical: 99.5% quality score, 5-min alert SLA. Financial reporting, customer-facing data. PagerDuty notification.
High: 99% quality score, 15-min alert SLA. Operational reports, team dashboards. Slack notification.
Medium: 95% quality score, daily digest. Internal analysis.
Low: weekly report. Exploratory data.
Escalation path: Test fails → team channel → on-call engineer → data quality lead. SLA breach documented in post-mortem with root cause.
Step 8: Data Quality Ecosystem Tools
re_data is an open-source framework that tracks row count, freshness, null rates, and distributions over time, building baselines for anomaly detection. Configure tables with YAML thresholds; auto-generates dbt tests from observed patterns. Use for automated baseline-driven quality monitoring without manual expectations.
dbt-audit-helper is a dbt package that compares two relations row-by-row on specified columns, reporting differences, missing rows, and mismatches. Essential for validating refactored dbt models produce identical results to originals.
ODD (Open Data Discovery) is an open-source observability platform ingesting metadata from data sources, tracking quality scores over time, with catalog and lineage. Integrates with dbt, Airflow, and Great Expectations. Use for centralized quality metric aggregation.
data-diff is an open-source tool for diffing tables across databases using checksum-based algorithms. Reports added, removed, and changed rows. Supports cross-database comparison (Postgres vs Snowflake). Use for migration validation, ETL QA, and source-target reconciliation.
# data-diff: compare tables across databases
data-diff --dbs postgresql://user@pg-host/db snowflake://user@sf-account/db \
--table public.orders public.orders \
--key order_id \
--columns status,total_amount,updated_at
Step 9: Extended Observability Integration
Combine re_data (baseline tracking) + dbt-audit-helper (migration validation) + ODD (centralized observability) + data-diff (cross-database comparison). Pipeline: re_data profiles new tables → generates expectations → feeds ODD quality metrics → data-diff validates ETL output → dbt-audit-helper validates refactoring → ODD alerts on score regression. This stack provides automated baselining, migration safety nets, and cross-system reconciliation without SaaS observability costs.
Rules
- Every critical table has a data contract
- Automated tests run on every pipeline execution
- Quality dimensions measured and trended weekly
- Freshness SLA defined per table, enforced by monitor
- Data contracts versioned alongside pipeline code
- Pipeline pauses on critical quality failure
- Quality score dashboard visible to data team and stakeholders
- No dashboard or report without quality metadata
- Soda checks for every production table
- Referential integrity verified across all related tables
References
- references/data-quality-automation.md — Data Quality Automation
- references/data-quality-ecosystem.md — Data Quality Ecosystem Tools
- references/data-quality-incident-management.md — Data Quality Incident Management
- references/data-quality-management.md — Data Quality Management
- references/data-quality-metrics.md — Data Quality Metrics
- references/data-quality-monitoring.md — Data Quality Monitoring
- references/ge-advanced-patterns.md — Great Expectations Advanced Patterns
- references/quality-automation.md — Quality Automation
- references/quality-dimensions.md — Data Quality Dimensions
- references/soda-check-examples.md — Soda Check Examples
Architecture Decision Trees
Data Quality Framework
├── Real-time quality checks?
│ ├── Yes → Great Expectations with Spark streaming / Deequ streaming
│ └── No → Batch quality checks (dbt tests, Soda)
├── Column-level quality needed?
│ ├── Yes → Great Expectations (rich expectation suite)
│ └── No → dbt generic tests (not null, unique, referential)
├── ML data quality?
│ ├── Yes → TensorFlow Data Validation (TFDV) for feature drift
│ └── No → Row-level checks (freshness, completeness)
└── Centralized quality platform?
├── Yes → Soda Cloud / Monte Carlo / Great Expectations Data Context
└── No → Per-pipeline inline checks
Decision criteria: Evaluate real-time requirements, team size, existing dbt usage, and ML data pipeline maturity.
Implementation Patterns
Great Expectations Checkpoint
# data_quality/ge_checkpoint.py
import great_expectations as ge
from great_expectations.core.batch import RuntimeBatchRequest
class DataQualityPipeline:
def __init__(self, context_path: str = "./great_expectations"):
self.context = ge.get_context(context_root_dir=context_path)
def run_checks(self, df, suite_name: str, datasource_name: str):
batch_request = RuntimeBatchRequest(
datasource_name=datasource_name,
data_connector_name="default_runtime_data_connector_name",
data_asset_name=suite_name,
runtime_parameters={"batch_data": df},
batch_identifiers={"default_identifier_name": "pipeline_run"},
)
validator = self.context.get_validator(
batch_request=batch_request,
expectation_suite_name=suite_name,
)
results = validator.validate()
validator.save_expectation_suite(discard_failed_expectations=False)
return results
dbt Quality Tests
# data_quality/dbt_tests.yml
version: 2
models:
- name: orders
columns:
- name: order_id
tests:
- unique
- not_null
- name: customer_id
tests:
- not_null
- relationships:
to: ref('customers')
field: customer_id
- name: total_amount
tests:
- dbt_expectations.expect_column_values_to_be_between:
min_value: 0
max_value: 100000
- name: silver_orders
tests:
- dbt_utils.expression_is_true:
expression: "order_date >= '2020-01-01'"
Production Considerations
- Alert routing: Route quality failures to domain team Slack channels; page on gold-level failures.
- Tiered severity: Bronze failures → log only; Silver failures → alert; Gold failures → page + block downstream.
- Quality SLAs: Define freshness (99% of tables within 24h), completeness (< 5% null rate), accuracy (< 1% error).
- Historical tracking: Store quality check results in a quality warehouse (table per dataset) for trend analysis.
- Data contracts: Enforce contracts via CI checks on schema change; block PR if contract compatibility fails.
- Root cause analysis: Link failures to upstream source changes via lineage; auto-file Jira tickets.
Anti-Patterns
| Anti-Pattern | Consequence | Solution |
|---|---|---|
| Testing everything equally | Noise, ignored alerts | Tier quality checks by criticality |
| No baseline for thresholds | False positives from unfamiliar data | Profile data first, set dynamic thresholds |
| Quality checks on production only | Bad data reaches consumers | Block at staging/Bronze layer |
| No observability integration | Alerts with no context | Link to catalog, lineage, dashboard |
| Ignoring data distribution drift | Static thresholds become obsolete | Periodic retraining of expectation baselines |
Performance Optimization
- Partitioned validation: Run quality checks per partition; only validate new/modified partitions.
- Sampling: Use stratified sampling for very large tables (10M+ rows); validate 100% only for critical columns.
- Parallel validation: Run table-level checks in parallel (Dask/Spark) for 5x faster quality runs.
- Caching expectations: Cache compiled expectations to avoid re-parsing suite definitions on each run.
- Incremental checks: Validate only changed rows using CDC streams instead of full table scans.
Security Considerations
- Quality metadata access: Restrict quality historical data access to data stewards and domain owners.
- PII in expectations: Never reference raw PII values in expectation parameters; use hashed references.
- Alert channels: Encrypt Slack webhook URLs; avoid including sensitive data values in alert messages.
- Schema validation: Validate quality results schema before writing to warehouse; reject malformed records.
- Audit trail: Log all quality configuration changes and threshold modifications for compliance.
Handoff
data-etl-pipeline for embedding quality checks into pipeline
data-bi-tools for displaying quality metadata on dashboards