Ferramenta de qualidade de dados costuma chegar como plataforma: um serviço para subir ao lado do cluster que já está com os dados na mão. E a saída costuma parar no veredito: "3.412 linhas falharam" diz que a tabela está ruim, não qual regra rejeitou qual linha, então a quarentena que ela escreve não dá para trabalhar.

Uma classe, Cola, com um registry de tipos de regra. Uma regra é um dict simples (a chave type mais os parâmetros daquele check), então o bloco de validação pode ser escrito à mão, gerado, ou guardado junto do pipeline que o executa. Thresholds são uma DSL pequena compartilhada por toda métrica: > 0, < 5%, between 10 and 20, <= 2h.

from sparquet_cola import Cola

cola = Cola()

for r in cola.run(df, [
    {"type": "row_count", "min": 1},
    {"type": "not_null", "columns": ["id", "email"]},
    {"type": "missing_percent", "column": "cpf", "must_be": "< 5%", "warn": "= 0"},
    {"type": "sql", "failed_rows": "SELECT * FROM _validation_df WHERE amount < 0"},
]):
    print(r)   # [FAIL] check: missing_percent(cpf) = 8 violates must_be (< 5%)

Toda regra row-level carrega um código: declarado, ou renderizado a partir da própria regra, sempre a mesma string para a mesma regra, porque essa string vai parar nos seus dados. O annotate escreve esse código ao lado de cada linha rejeitada, a partir dos predicados que o split já calcula, então não custa passada extra.

split = cola.split(df, [
    {"type": "not_null", "columns": ["id", "email"]},
    {"type": "range", "column": "age", "min": 1, "max": 99, "code": "AGE_RANGE"},
], annotate="dq_codes")

split.valid.write.format("delta").save(".../silver_ok")
split.invalid.write.format("delta").save(".../silver_quarantine")

split.invalid.select("id", "dq_codes").show(truncate=False)
# +----+-------------------+
# | id | dq_codes          |
# +----+-------------------+
# | 7  | [AGE_RANGE]       |
# | 8  | [not_null(email)] |
# +----+-------------------+

Cada target vira uma regra independente, com resultado próprio, código próprio e contribuição própria para a quarentena: um veredito único sobre duas colunas não diria qual delas quebrou. Formas ambíguas (lista de targets vazia, targets aninhado, code no pai) são recusadas em tempo de parse em vez de degradadas em silêncio.

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

# regex(document,^[0-9]{11}$)
# regex(document2,^[0-9]{12}$)

Até a v0.4.0 cada check disparava sua própria action, em geral mais de uma, então um bloco de uma dúzia de regras varria os dados uma dúzia de vezes. Agora o run junta as colunas de todo check que respondeu, emite um único df.agg(...) e devolve a cada check a sua fatia. Checks registrados de fora, que não têm como saber que o mecanismo existe, seguem com a própria action e seguem funcionando sem mudança.

O tests/test_run_batch_spark.py roda cada regra sozinha e de novo dentro de um bloco, comparando veredito, mensagem, failed_count, severity e metric_value campo a campo, para os dois caminhos não poderem divergir.

class NoFutureDateCheck(BaseCheck):
    def aggregations(self, df):
        return [F.count(F.when(F.col(self.params["column"]) > F.current_date(), F.lit(1)))]

    def from_aggregations(self, values):
        failed = int(values[0])
        if failed:
            return CheckResult("no_future_date", False, f"{failed} future dates", failed)
        return CheckResult("no_future_date", True)

• PyPI: pip install sparquet-cola, importado como sparquet_cola
• Entra em um job Spark, um notebook ou uma task do Airflow
• Documentado em inglês, português e espanhol
• É o motor por trás do bloco validations do Sparquet