Data quality in practice

Every pipeline should answer one question before it writes: is this data good enough to publish? Validations are how the answer becomes part of the file.

Start with four rules

For almost any dataset, these four catch most incidents:

{
  "validations": {
    "on_failure": "warn",
    "rules": [
      { "type": "row_count", "min": 1 },
      { "type": "not_null", "columns": ["id"] },
      { "type": "unique", "columns": ["id"] },
      { "type": "range", "column": "amount", "min": 0 }
    ]
  }
}

• row_count catches the silent empty load — an upstream feed that did not arrive, a filter that matched nothing.
• not_null on the key catches a broken join or a renamed source column.
• unique on the key catches the duplicated read and the fan-out join.
• range catches unit errors and sign flips.

Choose the failure policy deliberately

Policy	The pipeline	The report	When it is right
fail	aborts before writing	not written	The destination feeds something that must never see bad rows
warn	writes anyway	written	You want a history and a signal, not a stopped load
skip	writes anyway	written	The rules are still being calibrated

Caution

In fail mode the run aborts before the report is written, so a dashboard fed only by that table will never see hard failures. If you need every outcome recorded, use warn and let the orchestrator decide from result.validation_results.

result = fw.run("orders.json")
failed = [check for check in result.validation_results if not check.passed]
if failed:
    alert(f"{result.pipeline_name}: {len(failed)} rules failed")
    raise SystemExit(1)          # your policy, in your orchestrator

Persist the report

{
  "report": {
    "format": "delta",
    "path": "quality.pipeline_report",
    "mode": "append",
    "partition_by": ["pipeline"]
  }
}

Every run appends one row per rule: pipeline, rule_type, check_name, severity, passed, failed_count, metric_value, message, validated_at. Pointing every pipeline at the same table gives you a quality history for free, and three charts worth having:

• failure rate per rule over time — which check actually earns its keep
• failed_count trend per pipeline — degradation before it becomes an incident
• rules that never fail — either the data is solid or the rule is wrong

Rules for the things that actually break

Once the basics are in place, the valuable rules are domain rules. sql covers them: the query runs against _validation_df and must return a single boolean.

{
  "type": "sql",
  "query": "SELECT COUNT(*) = 0 FROM _validation_df WHERE ends_at < starts_at",
  "error_message": "Contracts ending before they start"
}

{
  "type": "sql",
  "query": "SELECT ABS(SUM(debit) - SUM(credit)) < 0.01 FROM _validation_df",
  "error_message": "Ledger does not balance"
}

{
  "type": "sql",
  "query": "SELECT COUNT(DISTINCT currency) = 1 FROM _validation_df",
  "error_message": "Mixed currencies in a single settlement batch"
}

Those three are worth more than a dozen null checks, because they encode what the business considers impossible.

Metric checks with a warning band

A metric rule measures one metric and compares it to a threshold, SODA-Core style — and it can warn before it fails. Every metric is a rule type: missing_percent, freshness, avg, row_count, … That two-level band is what lets a pipeline degrade visibly without stopping:

{ "type": "missing_percent", "name": "missing cpf",
  "column": "cpf", "warn": "= 0", "must_be": "< 1%" }

Zero missing values is the target (warn), anything up to 1% is tolerated (must_be), and past 1% the check fails. A warn is logged and lands in the report but never aborts the run, even under on_failure: "fail" — so you get an early signal without a false alarm.

The same shape covers most metric guards:

{ "type": "freshness", "column": "updated_at", "must_be": "< 1d" }
{ "type": "invalid_percent", "column": "email", "valid_format": "email", "must_be": "< 5%" }
{ "type": "avg", "column": "amount", "must_be": "between 10 and 100" }

For structural expectations — the columns and types a downstream consumer depends on — the schema rule fails fast when a source silently drops a column or changes a type:

{ "type": "schema", "required_columns": ["id", "amount"], "column_types": { "id": "bigint", "amount": "double" } }

See the validations reference for every metric and the full threshold DSL.

Validate, do not clean

// removes the rows
{ "type": "filter", "condition": "id IS NOT NULL" }

// counts them, changes nothing
{ "type": "not_null", "columns": ["id"] }

Both belong in most pipelines, and they answer different questions. A filter keeps the destination clean; a rule tells you how dirty the source was. Only the rule can tell you the feed is degrading.

Order matters

Validations run after every main transformation, so they measure the data you are about to write, not the raw source. To check the source instead, run a small pipeline whose only job is to read and validate it.

Guard a merge with a uniqueness rule

Delta and Iceberg raise multiple source rows matched when the incoming DataFrame has more than one row per merge key. A rule turns that runtime crash into a clear report:

{
  "validations": {
    "on_failure": "fail",
    "rules": [{ "type": "unique", "columns": ["order_id"] }]
  },
  "output": {
    "format": "delta",
    "path": "analytics.orders",
    "mode": "merge",
    "options": { "merge_keys": ["order_id"] }
  }
}

The same guard belongs on any incremental write keyed by a business id.

Two silvers: quarantine or pipeline outputs?

A recurring need: from one bronze table, produce a valid silver and an invalid silver. There are two honest ways to do it, and the right one depends on what “invalid” means.

• validations.outputs (quarantine) — the split is derived from the rules: a row is invalid when it fails a row-level rule (not_null, range, regex, unique, or a missing_*/invalid_* metric rule). One declarative definition of “invalid”, written next to the metrics, running apart from the main output — you don’t even need a main output. Best when invalid literally means “failed the data-quality rules”.
• Two pipeline outputs, each with its own filter transformation — the split is business logic you write by hand. Best when “invalid” is a domain rule unrelated to quality, or when the two branches need different shapes (columns, joins, payloads).

Rule of thumb: if you would otherwise write the same condition twice — once as a validation and once as a filter — use validations.outputs and let the checks be the single source of truth. If the branches diverge in meaning or shape, use pipeline outputs.

"validations": {
  "rules": [
    { "type": "not_null", "columns": ["id"] },
    { "type": "invalid_count", "column": "email", "valid_format": "email", "must_be": "= 0" }
  ],
  "outputs": {
    "valid":   { "format": "delta", "path": "silver.orders_ok",         "mode": "overwrite" },
    "invalid": { "format": "delta", "path": "silver.orders_quarantine", "mode": "overwrite" }
  }
}

For a targeted quarantine of a single rule’s offending rows, a sql rule with failed_rows plus its own output writes exactly those rows. The whole engine is the standalone sparquet_cola library — usable on its own in any Spark job.

A checklist

• row_count with a min on every pipeline
• not_null and unique on the business key
• range on money, quantities and dates
• regex on identifiers with a fixed shape (documents, codes)
• one sql for the invariant your domain would consider impossible
• warn plus a persisted report, unless the destination truly cannot tolerate bad rows
• a unique rule wherever a merge writes