Validations

Validations measure the DataFrame after every transformation and before any write. They never modify data — that separation is what makes the report trustworthy.

{
  "validations": {
    "on_failure": "warn",
    "report": { "format": "csv", "path": "/dq/orders", "mode": "append" },
    "rules": [
      { "type": "not_null", "columns": ["id", "customer_id"] },
      { "type": "unique", "columns": ["id"] },
      { "type": "range", "column": "quantity", "min": 1, "max": 10000 },
      { "type": "regex", "column": "email", "pattern": "^[^@\\s]+@[^@\\s]+\\.[a-z]{2,}$" },
      { "type": "row_count", "min": 1, "max": 5000000 },
      {
        "type": "sql",
        "query": "SELECT COUNT(*) = 0 FROM _validation_df WHERE revenue < 0",
        "error_message": "Negative revenue found"
      }
    ]
  }
}

Rules

not_null

Fails when any listed column contains nulls. The failure count is the number of offending rows.

{ "type": "not_null", "columns": ["id", "customer_id"] }

unique

Fails when the listed columns do not form a unique key. With several columns the uniqueness is over the combination, not each column separately.

{ "type": "unique", "columns": ["order_id", "line_number"] }

range

Bounds a numeric or date column. At least one of min / max is required; both are inclusive.

{ "type": "range", "column": "quantity", "min": 1, "max": 10000 }

Nulls pass this rule — pair it with not_null when the column is mandatory.

regex

Checks a string column against a pattern.

{ "type": "regex", "column": "document", "pattern": "^[0-9]{11}$" }

Note

Nulls count as failures here, unlike range. Remember JSON escaping: a backslash in the pattern is written \\.

row_count

Bounds the size of the DataFrame — the cheapest guard against a silently empty load.

{ "type": "row_count", "min": 1, "max": 5000000 }

min defaults to 0; max is optional.

sql

Anything the other rules do not express. The query runs against a temp view named _validation_df and must return a single boolean: true means the rule passed. Semantics are pass-when-true — write the invariant, not the violation.

{
  "type": "sql",
  "query": "SELECT COUNT(*) = 0 FROM _validation_df WHERE revenue < 0",
  "error_message": "Negative revenue found"
}

Use it for cross-column invariants (ends_at > starts_at), referential expectations inside the DataFrame, or business rules that only make sense together.

Two modes. Besides the boolean query, the sql rule accepts failed_rows — a query that returns the offending rows (SODA-style “failed rows”). The check fails if any come back, failed_count is their number, and an optional per-rule output writes exactly those rows to a sink:

{
  "type": "sql",
  "failed_rows": "SELECT * FROM _validation_df WHERE revenue < 0",
  "output": { "format": "delta", "path": "dq.negative_revenue", "mode": "overwrite" }
}

Use query when you only need pass/fail; use failed_rows when you want to see and keep the bad rows for debugging or a targeted quarantine.

Note

This rule was called custom_sql in earlier releases. It is now sql, with identical behavior. The whole validation engine is the standalone sparquet_cola library — same rule types.

Metric rules

Every metric is a rule type of its own: it measures the DataFrame (or one column) and compares the result to a threshold, in the SODA-Core style. Metric rules carry warn and fail severity levels, so one can pass and still raise a warning.

{ "type": "row_count", "must_be": "> 0" }

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

{ "type": "duplicate_count", "columns": ["id"], "must_be": "= 0" }

{ "type": "invalid_percent", "column": "email",
  "valid_format": "email", "must_be": "< 5%" }

{ "type": "avg", "column": "amount", "must_be": "between 10 and 100" }

{ "type": "freshness", "column": "updated_at", "must_be": "< 1d" }

Caution

Up to 0.5.x these were written as {"type": "check", "metric": "..."}. The wrapper existed only to hold the metric field — which is now the type itself — and 0.6.0 removed it with no alias: drop "type": "check" and promote metric to type. A report row now names the metric instead of saying check.

Metrics. row_count, distinct_count, missing_count / missing_percent, duplicate_count / duplicate_percent, invalid_count / invalid_percent, min, max, avg (mean), sum, stddev, and freshness (seconds since the newest value in column). Count-per-column metrics take column (or columns; the distinct/duplicate metrics use the whole set).

Threshold DSL (in must_be, the pass condition, and optional warn):

Form	Example
comparison	> 0, < 5, >= 100, <= 10, = 0, != 0
range	between 10 and 20, not between 1 and 2
percent suffix	< 5% (the % is cosmetic — the number is 5)
duration suffix	< 1d, <= 2h, > 30m (converted to seconds, for freshness)

Duration units are s m h d w. A bare number means equality (must_be: "0" ≡ = 0).

Column validity (for invalid_* metrics — a value counts as invalid when it is non-missing and fails every configured rule): valid_values, invalid_values, valid_format (named regexes: email, uuid, cpf, cnpj, date, timestamp, phone, url, ip, integer, decimal, boolean, alphanumeric, credit_card, …), valid_regex, valid_min / valid_max, valid_min_length / valid_max_length / valid_length. For missing_* metrics, missing_values treats extra sentinel strings (e.g. ["", "N/A"]) as missing on top of null.

targets — one entry, many rules

Any rule accepts targets: a list of objects that each become a rule of their own. Everything outside targets is a shared default; each target overrides what it needs.

{ "type": "regex", "targets": [
    { "column": "cpf",  "pattern": "^[0-9]{11}$" },
    { "column": "cnpj", "pattern": "^[0-9]{14}$" } ] }

{ "type": "range", "min": 1, "targets": [
    { "column": "id",  "max": 1000 },
    { "column": "age", "max": 120 } ] }

Independent is the point: every target gets its own result, its own report row and its own failure code, so the report says which column broke rather than giving one aggregate verdict. The range above is two rules — range(id,1,1000) and range(age,1,120) — and a quarantine can be scoped to just one of them.

Expansion happens when the config is parsed, so validations.rules is already flat by the time the rules run — and the report, which pairs rules with results by position, stays aligned.

A target cannot carry type (one entry is one rule type) or nested targets, and code / output belong inside each target rather than beside the list: every expanded rule would otherwise share one identifier or one destination. Each of these is refused when the config is parsed, instead of silently producing an ambiguous report.

examples/08_validacao_multi_alvo.json runs this end to end on local CSV: three rule entries, seven report rows.

schema

Asserts the shape of the DataFrame — which columns must exist, which must not, and their types.

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

column_types accepts type aliases: long → bigint, integer → int, and a decimal(p,s) expectation matches on the base decimal type regardless of precision/scale.

on_failure

What the engine does when at least one rule fails.

Mode	Pipeline	Report	Use it when
fail (default)	aborts before any write	not written	bad data must never reach the destination
warn	continues and writes	written	you want visibility without stopping the load
skip	continues and writes	written	the rules are informational for now

Caution

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

The mode is compared case-sensitively — write it lowercase.

Severity

Metric rules add a level between pass and fail. A rule that clears must_be but trips its optional warn threshold has warn severity: it is logged and reported, but it counts as passing — a warn never aborts the run, even under on_failure: "fail". Only a real failure (must_be violated, or any other rule failing) does.

Each result carries the severity alongside the classic fields:

Field	Meaning
severity	pass, warn or fail — derived from passed when a rule does not set it
metric_value	the number a metric rule measured (None for rules that produce no metric)
check_name	the rule’s optional name, for readable reports

Because a warn keeps passed = True, an orchestrator filtering on not check.passed sees only hard failures; inspect check.severity == "warn" to surface warnings too.

The report

report accepts a full output configuration — any format, any mode, even its own transformations. One row per rule:

Column	Meaning
pipeline	the pipeline name
rule_type	validator type
check_name	the check’s name, when set (empty otherwise)
severity	pass, warn or fail
passed	boolean (true for pass and warn)
failed_count	offending rows (0 when passed)
metric_value	the number a metric rule measured (null otherwise)
message	human-readable failure detail
validated_at	timestamp of the check

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

Appending every run into one table gives you a quality history you can chart: failure rate per rule, per pipeline, over time.

Row-level quarantine (validations.outputs)

Where report persists metrics, validations.outputs routes the rows themselves — apart from the main output(s). Keys valid and invalid, each a full output sink:

"validations": {
  "on_failure": "warn",
  "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" }
  }
}

A row is invalid when it violates any row-level rule — not_null, range, regex, unique, and the missing_* / invalid_* metric rules. Aggregate rules (row_count, avg, freshness, duplicate_count, schema, boolean sql) describe the table, not individual rows, so they do not contribute to the split.

A copy, not a split

validations.outputs does not divert rows. The pipeline writes these sinks and then writes its own output/outputs from the same, complete DataFrame — invalid rows included. A pipeline with quarantine writes every bad row twice.

So it audits: which rows broke a rule, kept apart for inspection, without changing the flow.

To route — only the good rows reaching the curated table — use the pipeline: a filter before each destination. You restate the condition in SQL (keep it in step with the rules), and in exchange the main destination carries exactly what you meant.

Reading the result in Python

result = fw.run("orders.json")

for check in result.validation_results:
    status = "ok" if check.passed else f"{check.failed_count} rows"
    print(f"{check.rule_type:12} {status:>12}  {check.message}")

if any(not check.passed for check in result.validation_results):
    notify_data_owner(result.pipeline_name)

Choosing between a rule and a filter

Intent	Right tool
The row must not reach the destination	filter in transformations
I need to know how many bad rows arrived	a validation rule
Bad data means the load is wrong	a rule with on_failure: "fail"
Bad data is expected but must be tracked	a rule with warn plus a report

Custom validators

from sparquet.validation.base import BaseValidator, ValidationResult
import pyspark.sql.functions as F

class NoFutureDateValidator(BaseValidator):
    def validate(self, df):
        column = self.rule.params["column"]
        failed = df.filter(F.col(column) > F.current_date()).count()
        if failed:
            return ValidationResult("no_future_date", False, f"{failed} future dates", failed)
        return ValidationResult("no_future_date", True)

fw.register_validator("no_future_date", NoFutureDateValidator)

{ "type": "no_future_date", "column": "ordered_at" }

See Extending.