Data Validator

---
name: data-validator
description: |
  Data validation and quality expertise covering Great Expectations patterns, schema validation, statistical validation, referential integrity checks, data profiling, anomaly detection, data contracts, quality scoring, and automated testing strategies for ensuring data reliability throughout the pipeline.
  Use when the user asks about data validator, data validator best practices, or needs guidance on data validator implementation.
  Do NOT use when the user needs a different specialized skill or is asking about an unrelated technology domain.
license: Apache-2.0
metadata:
  author: foundry-skills
  version: "1.0.0"
  tags: "data-science sql testing"
  category: "data-engineering"
  subcategory: "pipelines-etl"
  depends: ""
  disclaimer: "none"
  difficulty: "intermediate"
---

# Data Validator

## Overview

Data quality is the foundation upon which all downstream analytics, ML models, and business decisions depend. This skill covers tools, techniques, and patterns for validating data at every stage of the pipeline.

## Great Expectations Patterns

### Setup and Expectation Suites

```python
import great_expectations as gx

context = gx.get_context()
datasource = context.sources.add_pandas("pandas_datasource")

# Build expectation suite
suite = context.add_expectation_suite("customer_quality_suite")

# Table-level
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(min_value=10000, max_value=10000000))
suite.add_expectation(gx.expectations.ExpectTableColumnCountToEqual(value=15))

# Column-level: customer_id
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="customer_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToMatchRegex(
    column="customer_id", regex=r"^CUS-[A-Z0-9]{8}$"
))

# Column-level: email (allow 1% non-matching for legacy data)
suite.add_expectation(gx.expectations.ExpectColumnValuesToMatchRegex(
    column="email",
    regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
    mostly=0.99
))

# Numeric column: revenue
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="lifetime_revenue", min_value=0, max_value=10000000, mostly=0.999
))

# Categorical column
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(
    column="status", value_set=["active", "inactive", "suspended", "pending"]
))
```

### Running Validations in Pipelines

```python
def validate_data(**context):
    ge_context = gx.get_context()
    checkpoint = ge_context.add_or_update_checkpoint(
        name="customer_checkpoint",
        validations=[{
            "batch_request": {"datasource_name": "warehouse", "data_asset_name": "dim_customer"},
            "expectation_suite_name": "customer_quality_suite",
        }],
        action_list=[
            {"name": "store_validation_result", "action": {"class_name": "StoreValidationResultAction"}},
            {"name": "update_data_docs", "action": {"class_name": "UpdateDataDocsAction"}},
        ]
    )
    result = checkpoint.run()
    if not result.success:
        raise ValueError("Data quality check failed")
```

## Schema Validation

### Schema Contract (YAML)

```yaml
name: dim_customer
version: "2.1"
owner: data-engineering
sla_hours: 6
allow_extra_columns: false
min_rows: 50000

columns:
  - name: customer_id
    type: string
    nullable: false
    unique: true
    pattern: "^CUS-[A-Z0-9]{8}$"
  - name: email
    type: string
    nullable: true
    max_length: 255
  - name: lifetime_revenue
    type: float
    nullable: false
    min_value: 0
  - name: status
    type: string
    nullable: false
    allowed_values: ["active", "inactive", "suspended", "pending"]
```

### Schema Drift Detection

```python
class SchemaDriftDetector:
    def __init__(self, schema_store):
        self.store = schema_store

    def detect_drift(self, table_name: str, current_df) -> dict:
        previous_schema = self.store.get_schema(table_name)
        current_schema = self._extract_schema(current_df)
        if previous_schema is None:
            self.store.save_schema(table_name, current_schema)
            return {'status': 'new_table', 'changes': []}

        changes = []
        prev_cols = {c['name']: c for c in previous_schema['columns']}
        curr_cols = {c['name']: c for c in current_schema['columns']}

        for name in set(curr_cols) - set(prev_cols):
            changes.append({'type': 'column_added', 'column': name})
        for name in set(prev_cols) - set(curr_cols):
            changes.append({'type': 'column_removed', 'column': name})
        for name in set(prev_cols) & set(curr_cols):
            if prev_cols[name]['dtype'] != curr_cols[name]['dtype']:
                changes.append({'type': 'type_changed', 'column': name,
                    'from': prev_cols[name]['dtype'], 'to': curr_cols[name]['dtype']})

        if changes:
            self.store.save_schema(table_name, current_schema)
        return {'status': 'drift_detected' if changes else 'no_change', 'changes': changes}
```

## Statistical Validation

```python
from scipy import stats
import numpy as np

class StatisticalValidator:
    @staticmethod
    def detect_outliers_iqr(series, multiplier=1.5):
        Q1, Q3 = series.quantile(0.25), series.quantile(0.75)
        IQR = Q3 - Q1
        lower, upper = Q1 - multiplier * IQR, Q3 + multiplier * IQR
        outliers = series[(series < lower) | (series > upper)]
        return {'count': len(outliers), 'percentage': len(outliers) / len(series) * 100,
                'lower_bound': lower, 'upper_bound': upper}

    @staticmethod
    def compare_distributions(current, historical, significance=0.05):
        stat, p_value = stats.ks_2samp(current.dropna(), historical.dropna())
        return {'test': 'kolmogorov_smirnov', 'statistic': stat, 'p_value': p_value,
                'significant_drift': p_value < significance}

    @staticmethod
    def validate_proportions(series, expected_proportions):
        observed = series.value_counts(normalize=True).sort_index()
        expected = pd.Series(expected_proportions).sort_index()
        all_cats = sorted(set(observed.index) | set(expected.index))
        observed = observed.reindex(all_cats, fill_value=0)
        expected = expected.reindex(all_cats, fill_value=0)
        stat, p_value = stats.chisquare(observed * len(series), expected * len(series))
        return {'test': 'chi_squared', 'p_value': p_value, 'significant_difference': p_value < 0.05}
```

## Referential Integrity

```python
class ReferentialIntegrityChecker:
    def __init__(self, engine):
        self.engine = engine

    def check_fk(self, child_table, child_col, parent_table, parent_col):
        query = f"""
            SELECT COUNT(*) AS orphan_count
            FROM {child_table} c
            LEFT JOIN {parent_table} p ON c.{child_col} = p.{parent_col}
            WHERE p.{parent_col} IS NULL AND c.{child_col} IS NOT NULL
        """
        result = pd.read_sql(query, self.engine).iloc[0]
        return {'child_table': child_table, 'parent_table': parent_table,
                'orphan_count': int(result['orphan_count']),
                'passed': result['orphan_count'] == 0}
```

## Anomaly Detection

```python
class AnomalyDetector:
    def detect_volume_anomaly(self, current_count, historical_counts, z_threshold=3.0):
        mean, std = np.mean(historical_counts), np.std(historical_counts)
        if std == 0:
            return {'is_anomaly': current_count != mean}
        z_score = (current_count - mean) / std
        return {'is_anomaly': abs(z_score) > z_threshold, 'z_score': z_score,
                'expected_range': (mean - z_threshold * std, mean + z_threshold * std)}

    def detect_freshness_anomaly(self, latest_timestamp, expected_frequency_hours):
        from datetime import datetime, timezone
        age_hours = (datetime.now(timezone.utc) - latest_timestamp).total_seconds() / 3600
        return {'is_stale': age_hours > expected_frequency_hours * 2, 'age_hours': age_hours}
```

## Data Quality Scoring

```python
class DataQualityScorer:
    def __init__(self):
        self.dimensions = {
            'completeness': 0.25, 'uniqueness': 0.15, 'validity': 0.25,
            'consistency': 0.15, 'timeliness': 0.10, 'accuracy': 0.10,
        }

    def score(self, validation_results: dict) -> dict:
        scores = {}
        weighted_total = 0
        for dimension, weight in self.dimensions.items():
            if dimension in validation_results:
                dim_score = validation_results[dimension]
                scores[dimension] = {'score': dim_score, 'weight': weight,
                    'grade': 'A' if dim_score >= 95 else 'B' if dim_score >= 85 else 'C' if dim_score >= 70 else 'D' if dim_score >= 50 else 'F'}
                weighted_total += dim_score * weight
        overall = weighted_total / sum(self.dimensions[d] for d in scores) if scores else 0
        return {'overall_score': round(overall, 2), 'dimensions': scores}
```

## Data Contracts

```yaml
contract:
  name: dim_customer
  version: "3.0"
  owner: data-engineering
  consumers: [analytics, marketing-ml, customer-success]

  schema:
    fields:
      - { name: customer_id, type: string, required: true, unique: true }
      - { name: email, type: string, required: false, pii: true }
      - { name: lifetime_revenue, type: "decimal(12,2)", required: true, min: 0 }

  quality:
    freshness: { max_age_hours: 24, field: updated_at }
    volume: { min_rows: 50000, max_row_change_pct: 20 }
    completeness: { email: 95, phone: 80 }

  sla:
    availability: 99.9
    update_frequency: daily

  breaking_changes:
    notification_days: 14
    channels: [{ slack: "#data-contracts" }]
```

## Automated Testing: dbt Integration

```sql
-- tests/generic/test_revenue_positive.sql
{% test positive_revenue(model, column_name) %}
SELECT * FROM {{ model }} WHERE {{ column_name }} < 0
{% endtest %}

-- tests/singular/assert_revenue_reconciliation.sql
WITH warehouse AS (
    SELECT SUM(net_amount) AS total FROM {{ ref('fct_orders') }}
    WHERE order_date = '{{ var("check_date") }}'
),
source AS (
    SELECT SUM(amount) AS total FROM {{ source('stripe', 'charges') }}
    WHERE DATE(created) = '{{ var("check_date") }}'
)
SELECT * FROM warehouse w CROSS JOIN source s
WHERE ABS(w.total - s.total) / GREATEST(s.total, 1) > 0.01
```

## Validation Decision Framework

| Stage | What to Validate | Failure Action |
|-------|-----------------|----------------|
| Ingestion | Schema, encoding, row count | Reject file, alert source team |
| Staging | Types, nulls, basic ranges | Quarantine records, log to DLQ |
| Transform | Business rules, referential integrity | Block downstream, alert owner |
| Load | Row counts match, no duplicates | Rollback, retry with investigation |
| Serving | Freshness, availability, SLA | Alert consumers, serve stale with warning |

## When to Use

**Use this skill when:**
- Designing or implementing data validator solutions
- Reviewing or improving existing data validator approaches
- Making architectural or implementation decisions about data validator
- Learning data validator patterns and best practices
- Troubleshooting data validator-related issues

**Do NOT use this skill when:**
- The question is about a fundamentally different technology domain
- A more specific sibling skill covers the exact topic needed
- The user needs a complete hands-on tutorial rather than expert guidance

## Output Format

```markdown
# Data Validator Analysis

## Context Assessment
[Situation summary and constraints]

## Recommended Approach
[Primary recommendation with rationale]

## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]

## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]

## Next Steps
- [Immediate action item]
- [Follow-up action item]
```

## Example

**Input:** "Help me implement data validator for a medium-scale production application"

**Output:** A structured analysis covering current state assessment, recommended data validator approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.

## Edge Cases

- **Legacy system integration:** When data validator must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
- **Scale mismatch:** When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
- **Team skill gaps:** When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
- **Conflicting requirements:** When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities