sparquet_cola — data quality library

sparquet_cola is the data-quality layer of Sparquet, extracted as a library you can use on its own. Cola (“glue” in Portuguese) is the layer that glues quality onto your DataFrames: SODA-Core-style metric checks, free-form SQL rules, schema assertions, and a valid/invalid split (quarantine) — all on top of PySpark.

It depends only on pyspark. That is deliberate: you can drop it into any Spark job, notebook or Airflow task without pulling in the rest of the framework. Inside Sparquet it is the engine behind the validations block — the rule type strings are identical, so what you learn here transfers directly to the JSON.

Install and import

sparquet-cola is its own package (repo). Install it standalone, or get it automatically as a dependency of sparquet:

pip install sparquet-cola   # standalone (pyspark only)
pip install sparquet        # the framework — brings sparquet-cola along

from sparquet_cola import Cola

The public names are Cola, ColaSplit, CheckResult, and the individual check classes if you want to reference them directly.

The Cola API

One class does everything. It holds a registry of check types and exposes four members:

from sparquet_cola import Cola

cola = Cola()

# run(df, rules) -> list[CheckResult]
results = cola.run(df, [
    {"type": "row_count", "min": 1},
    {"type": "not_null", "columns": ["id"]},
    {"type": "missing_percent", "column": "cpf", "must_be": "< 5%"},
])
for r in results:
    print(r)   # [FAIL] check: missing_percent(cpf) = 8 violates must_be (< 5%)

# split(df, rules) -> ColaSplit(valid, invalid)
split = cola.split(df, [{"type": "not_null", "columns": ["id"]}])
split.valid.write.format("delta").save(".../silver_ok")
split.invalid.write.format("delta").save(".../silver_quarantine")

# register(name, cls) -> add a custom check available by `type`
cola.register("no_future_date", NoFutureDateCheck)

# available -> the registered check types
print(cola.available)
# ['check', 'not_null', 'range', 'regex', 'row_count', 'schema', 'sql', 'unique']

Member	Signature	Returns
run	run(df, rules)	a CheckResult per rule, in order
split	split(df, rules)	a ColaSplit(valid, invalid) of two DataFrames
register	register(name, cls)	registers a custom check under a type
available	property	sorted list of registered check types

Each rule is a plain dict with a type key plus that check’s parameters — the same shape as an entry in the JSON validations.rules. A CheckResult carries:

Field	Meaning
rule_type	the check’s type
passed	True / False
severity	pass, warn or fail (SODA-style)
message	human-readable detail
failed_count	offending rows / problems (0 when passed)
metric_value	the number a metric rule measured (None otherwise)
check_name	the optional name you gave the rule
failed_rows	for a sql failed_rows check, the DataFrame of bad rows

Note

run never raises on a failed check — it returns the result and lets you decide. It only raises for a genuinely malformed rule (unknown type, a missing required parameter, an invalid threshold or metric name).

The checks

not_null

Fails when any listed column contains a NULL. failed_count is the total NULLs across all listed columns.

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

• columns — always a list, even for one column. Empty string '' and NaN are not NULL and pass.

This is a row-level check: it feeds the valid/invalid split.

unique

Fails when the combination of the listed columns is not unique. Uniqueness is over the tuple of columns, not each column separately.

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

• columns — the composite key. failed_count is the number of excess rows (total - distinct). Spark treats NULLs as equal, so many rows with a NULL key are reported as duplicates.

Row-level: contributes to the split (a row whose key appears more than once is invalid).

range

Fails when a numeric or date column falls outside the inclusive [min, max] interval.

{"type": "range", "column": "amount", "min": 0, "max": 10000}

• column (singular) — the column to bound.
• min / max — at least one is required; both bounds are inclusive. NULLs pass — pair with not_null when the column is mandatory.

Row-level: contributes to the split.

regex

Fails when a string column does not match a pattern (rlike, a partial match — anchor with ^...$ for a full-string match).

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

• column — the string column.
• pattern — a Java/Spark regex. NULLs count as failures here (the opposite of range).

Row-level: contributes to the split.

row_count

Table-level guard on the size of the DataFrame. Fails when count < min or, when given, count > max.

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

• min — inclusive lower bound, defaults to 0.
• max — optional inclusive upper bound.

Aggregate check — it does not contribute to the split (you cannot blame a row for the table’s size).

sql

Free-form SQL over the temp view _validation_df, which sparquet_cola registers for you. Two modes:

Invariant mode (query) — pass-when-true. The query returns a single boolean; the check passes when it is true. Write the invariant, not the violation.

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

Failed-rows mode (failed_rows) — the query returns the offending rows. The check fails if any come back, and failed_count is their count. The bad rows are attached to result.failed_rows as a DataFrame, so you can route them to a sink.

{"type": "sql",
 "failed_rows": "SELECT * FROM _validation_df WHERE amount < 0",
 "error_message": "Negative amounts found",
 "output": {"format": "delta", "path": "dq.negative_amounts", "mode": "overwrite"}}

• query or failed_rows — use one, not both.
• error_message — shown when the check fails (default "SQL validation failed").
• output — an optional full output config { format, path, mode, ... }. Only meaningful with failed_rows; the framework writes the bad rows there (see validations reference). Used directly with Cola, the DataFrame is on result.failed_rows for you to write yourself.
• view_name — optional; defaults to _validation_df.

Metric rules (SODA-style)

A single metric measured over the DataFrame (or a column) and compared to a threshold, with optional warn and fail levels. This is the big-data take on a SODA Core check.

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

• metric (required) — one of:

  Metric	Measures
  row_count	total rows (no column needed)
  distinct_count	distinct rows over the df or columns
  missing_count / missing_percent	NULLs (plus missing_values) in column
  duplicate_count / duplicate_percent	duplicate rows over the df or columns
  invalid_count / invalid_percent	present values in column that fail the validity config
  min / max / avg (mean) / sum / stddev	numeric aggregate over column
  freshness	seconds since the newest value in column

• must_be (required) — the pass condition (the threshold DSL, below).

• warn — optional softer threshold. A metric that passes must_be but fails warn gets severity warn: logged and reported but not a failure (it never aborts the framework’s pipeline, even under on_failure: "fail").

• column / columns — the target column(s) for column-scoped metrics.

• name — a label carried into the result for readable reports.

Threshold DSL (used in must_be and 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). Percent metrics are 0–100 over the row count.

Validity config (for invalid_count / invalid_percent — a value is invalid when it is present and fails every configured rule):

Key	Effect
valid_values	value must be in this list
invalid_values	value must not be in this list
valid_format	a named regex (below)
valid_regex	a custom regex
valid_min / valid_max	numeric bounds
valid_min_length / valid_max_length / valid_length	string length bounds

Named valid_format values: email, uuid, phone, integer, decimal, number, percentage, date, timestamp, ip, ipv4, url, boolean, alphanumeric, credit_card, cpf, cnpj.

For missing_* metrics, missing_values treats extra sentinel strings (e.g. ["", "N/A"]) as missing on top of null.

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

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

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

Row-level for the split only for missing_* and invalid_* metrics (those can point at a row); aggregate metrics (row_count, avg, freshness, duplicate_*…) do not contribute to the split.

schema

A basic data contract: which columns must exist, which must not, and their types. Fails listing every problem in a single message.

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

• required_columns — columns that must exist.
• forbidden_columns — columns that must not exist (e.g. a leaked internal field).
• column_types — a map of column → expected Spark type. Matching is case-insensitive with aliases (long → bigint, integer → int, str → string); decimal(p,s) matches on the base decimal type regardless of precision/scale.

Aggregate check — it does not contribute to the split.

The valid/invalid split (quarantine)

split(df, rules) divides a DataFrame into valid and invalid using the row-level checks. A row is invalid when it violates any check that can point at rows — not_null, range, regex, unique, and the missing_* / invalid_* metric rules. Aggregate checks (row_count, avg, freshness, duplicate_count, schema, sql…) are ignored by the split. If no rule is row-level, everything is valid.

from sparquet_cola import Cola

cola = Cola()

split = cola.split(orders, [
    {"type": "not_null", "columns": ["id", "customer_id"]},
    {"type": "range", "column": "amount", "min": 0},
    {"type": "invalid_count", "column": "email",
     "valid_format": "email", "must_be": "= 0"},
])

split.valid.write.format("delta").mode("append").save("lake/silver/orders")
split.invalid.write.format("delta").mode("append").save("lake/quarantine/orders")

print(f"{split.valid.count()} good, {split.invalid.count()} quarantined")

Every row that is NULL in id/customer_id, negative in amount, or carries a non-empty invalid e-mail lands in invalid; the rest in valid. The two DataFrames partition the input — no row is lost or duplicated.

Note

A NULL comparison never invalidates a row on its own — e.g. a NULL amount is out of range only via the not_null rule, not the range rule. That keeps the split predictable: each rule owns exactly the rows it can name.

Custom checks

Subclass BaseCheck, implement run(df) -> CheckResult, and optionally violation(df) (a boolean Spark Column, True for offending rows) to join the split. Register it under a type:

from pyspark.sql import functions as F
from sparquet_cola import Cola
from sparquet_cola.checks import BaseCheck, CheckResult

class NoFutureDateCheck(BaseCheck):
    def run(self, df):
        column = self.params["column"]
        failed = df.filter(F.col(column) > F.current_date()).count()
        if failed:
            return CheckResult("no_future_date", False, f"{failed} future dates", failed)
        return CheckResult("no_future_date", True)

    def violation(self, df):
        return F.col(self.params["column"]) > F.current_date()

cola = Cola()
cola.register("no_future_date", NoFutureDateCheck)
cola.run(df, [{"type": "no_future_date", "column": "ordered_at"}])

Inside the framework

The validations block of a pipeline JSON runs on exactly this engine — the same rule type strings, the same threshold DSL, the same validity config. The framework adds report persistence, the on_failure policy, and row-level quarantine via validations.outputs (valid / invalid sinks). See the validations reference and the data-quality guide.