---
description: Data Quality
alwaysApply: false
---

# Data Quality

Patterns for validating, monitoring, and ensuring data quality.

## Quality Dimensions

- **Completeness** — required data is present (no unexpected nulls)
- **Accuracy** — data reflects reality (values in expected ranges)
- **Consistency** — data agrees across systems (totals match line items)
- **Timeliness** — data is fresh enough (within SLA)
- **Uniqueness** — no unwanted duplicates (primary keys are unique)
- **Validity** — data conforms to business rules (email matches regex)

## Validation Patterns

- **Schema validation** — enforce expected column names, types, and nullability before processing
- **Null checks** — assert zero nulls in required columns; fail pipeline on violation
- **Uniqueness checks** — `groupBy(keys).count().filter("count > 1")` must be empty
- **Range checks** — numeric values within `[min, max]`; flag or reject out-of-range
- **Referential integrity** — left-anti join to find orphan foreign keys

```python
def check_required_columns(df, required):
    for col in required:
        if df.filter(F.col(col).isNull()).count() > 0:
            raise DataQualityError(f"Nulls in required column: {col}")
    return df
```

## Severity Levels

- **Critical** — pipeline must fail (null PKs, duplicates on unique keys)
- **Warning** — log and continue (invalid emails, future dates); emit metrics for trending

## Freshness Monitoring

- Define SLA per table (e.g., orders: 2h, daily_sales: 24h)
- Compare `MAX(timestamp_column)` against current time
- Alert on SLA breaches with severity matching business impact

## Anomaly Detection

- **Volume** — compare today's record count against rolling 30-day mean ± 3 std devs
- **Distribution drift** — detect shifts in categorical value distributions (KL divergence)
- Zero-count partitions or sudden spikes both warrant alerts

## Quarantine Pattern

Separate good and bad records; route failures to a quarantine table for investigation.

```python
good = df.filter("order_id IS NOT NULL AND total_amount >= 0")
bad = df.subtract(good)
if bad.count() > 0:
    bad.withColumn("_reason", F.lit("validation_failed")) \
       .write.mode("append").saveAsTable("quarantine.orders")
```

## DBT Tests

```yaml
columns:
  - name: order_id
    tests: [not_null, unique]
  - name: status
    tests: [{accepted_values: {values: [pending, confirmed, shipped]}}]
```

## Anti-Patterns

- Silently dropping bad records without logging or quarantine
- Only validating in development; skipping checks in production
- No freshness monitoring — stale data serves silently