Data Quality Checks skill for Claude Code

Data quality work is done when each important dataset has checks for the failure modes that have actually hurt (or plausibly will), blocking checks stop bad data before it is published, warnings reach an owner, and every check has a threshold someone chose on purpose rather than a default.

Ground rules

• Start from consumers: which numbers would be wrong, and which failure would cause it? Check those first. Fifty generic not-null tests are less useful than one volume check.
• Validate at boundaries: on ingestion (contract with the producer), after transformation, and before publishing.
• Every check has a severity: error blocks publishing, warn notifies.
• Collect all failures in one run (lazy validation) so a fix round covers everything.
• Bad rows go to a quarantine table with the reason, not silently dropped.
• Use the tool the project already runs; add a new framework only with a reason.

Workflow

- [ ] 1. List datasets, consumers, owners, and past incidents
- [ ] 2. Profile current data to set realistic thresholds
- [ ] 3. Choose checks per dataset (table below) and severities
- [ ] 4. Implement in the project's tool
- [ ] 5. Wire into the pipeline: block on error, alert on warn
- [ ] 6. Prove each check fails on bad data (inject a fault)
- [ ] 7. Document checks and owners

3. Check catalog

Dimension	Example check	Typical severity
Schema	Expected columns and types present; no unexpected columns (strict)	error
Uniqueness	Primary/grain key unique	error
Completeness	Key and required fields not null; null share of optional field < X%	error / warn
Validity	Ranges (amount >= 0), enums, regex (emails, ISO country), date bounds (not in future)	error / warn
Referential integrity	Every customer_id exists in customers	warn (late-arriving dims) or error
Freshness	max(loaded_at) within N hours of now	error for SLAs
Volume	Row count within expected band vs trailing history	warn, error if zero
Distribution	Mean/sum/category mix within band vs history	warn
Reconciliation	Target totals equal source totals for the partition	error for finance data
Consistency	total = sum(line_items); end_date >= start_date	error

2. Profile to set thresholds

-- Daily volume history: use to set the volume band
select date(created_at) as day, count(*) as n
from analytics.orders
where created_at >= current_date - interval '56 days'
group by 1 order by 1;

-- Null share per column
select
  avg(case when email is null then 1.0 else 0 end) as email_null_share,
  avg(case when country is null then 1.0 else 0 end) as country_null_share
from analytics.customers;

Set bands from history (e.g. weekday-aware mean ± 3 standard deviations, or ± 30%) and revisit after a few weeks of alerts.

4. Implementations

Pandera (DataFrames in Python pipelines):

import pandas as pd
import pandera.pandas as pa      # older pandera: import pandera as pa
from pandera.typing import Series

class Orders(pa.DataFrameModel):
    order_id: Series[str] = pa.Field(unique=True)
    customer_id: Series[str] = pa.Field(nullable=False)
    amount: Series[float] = pa.Field(ge=0, le=100_000)
    status: Series[str] = pa.Field(isin=["pending", "paid", "refunded"])
    created_at: Series[pd.Timestamp]

    @pa.dataframe_check
    def not_in_future(cls, df: pd.DataFrame) -> Series[bool]:
        return df["created_at"] <= pd.Timestamp.now(tz=df["created_at"].dt.tz)

    class Config:
        strict = True      # unexpected columns fail
        coerce = True

try:
    Orders.validate(df, lazy=True)
except pa.errors.SchemaErrors as exc:
    failures = exc.failure_cases          # DataFrame: column, check, failure_case, index
    failures.to_parquet("quarantine/orders_failures.parquet")
    raise

Great Expectations (GX Core 1.x):

import great_expectations as gx

context = gx.get_context()
source = context.data_sources.add_pandas("pandas")
asset = source.add_dataframe_asset(name="orders")
batch_def = asset.add_batch_definition_whole_dataframe("orders_batch")
suite = context.suites.add(gx.ExpectationSuite(name="orders_suite"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(column="amount", min_value=0))
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(min_value=10_000, max_value=80_000))
result = batch_def.get_batch(batch_parameters={"dataframe": df}).validate(suite)
if not result.success:
    raise RuntimeError(result.describe())

GX’s API changed substantially at 1.0; do not mix 0.x tutorials (ge.from_pandas, DataContext().run_checkpoint) into 1.x code.

dbt (warehouse models):

models:
  - name: fct_orders
    data_tests:
      - dbt_utils.recency:
          datepart: hour
          field: created_at
          interval: 6
    columns:
      - name: order_id
        data_tests: [unique, not_null]
      - name: amount_usd
        data_tests:
          - dbt_utils.accepted_range:
              min_value: 0
              config: { severity: error }
      - name: country
        data_tests:
          - not_null:
              config: { severity: error, warn_if: ">100", error_if: ">1000" }   # warn above 100, fail above 1000

Soda (SodaCL):

checks for orders:
  - row_count > 0
  - missing_count(order_id) = 0
  - duplicate_count(order_id) = 0
  - invalid_count(status) = 0:
      valid values: [pending, paid, refunded]
  - freshness(created_at) < 6h

Plain SQL (any warehouse; a check passes when it returns zero rows):

-- Volume anomaly: today vs same weekday over the previous 8 weeks
with hist as (
  select date(created_at) d, count(*) n
  from analytics.orders
  where created_at >= current_date - interval '57 days'
  group by 1
), stats as (
  select avg(n) mu, stddev(n) sd from hist
  where extract(dow from d) = extract(dow from current_date - 1) and d < current_date - 1
)
select h.n, s.mu, s.sd
from hist h cross join stats s
where h.d = current_date - 1 and abs(h.n - s.mu) > 3 * s.sd;

5. Wire into the pipeline

Severity	Pipeline behavior
error	Stop before publish (fail task / do not swap partition); alert owner
warn	Publish; notify channel with failing counts and sample rows
quarantine	Publish good rows; write bad rows + reason to <table>_quarantine; alert if share > threshold

Alerts include: dataset, partition, check name, observed vs expected, sample failing rows (redacted), runbook link, owner.

6. Prove the checks work

For each new check, run it once against deliberately broken data (duplicate a key, null a column, drop a day’s rows, shift timestamps) and confirm it fails with a readable message. A check that has never failed is unverified.

Gotchas

• Row-level checks on billions of rows are expensive; check the new partition, not the whole table.
• Freshness checks based on created_at fail on legitimately quiet periods; prefer a load timestamp.
• Volume bands that ignore weekly seasonality alert every weekend.
• Referential checks fail on late-arriving dimensions; use warn or a grace window.
• strict=True schemas break when producers add columns; agree a contract for additions.
• Alert fatigue kills data quality programs; tune or delete checks that fire without action.
• Floating-point reconciliation needs a tolerance (abs(a - b) < 0.01), not equality.

Output

## Data quality: analytics.orders (owner: data-platform)
| Check | Tool | Severity | Threshold | Status |
|---|---|---|---|---|
| order_id unique, not null | dbt | error | 0 failures | pass |
| freshness(loaded_at) | dbt recency | error | < 6 h | pass |
| daily volume vs same weekday | SQL | warn | ±3 sd (8 wks) | pass |
| amount >= 0 | dbt accepted_range | error | 0 failures | pass |
| status in enum | Soda | warn | 0 invalid | pass |
Fault injection: each check failed as expected on a corrupted staging copy.