dq-engine

Framework PySpark para validacao e tratamento de qualidade de dados.

O que faz

• Resolve rule_set por nome de coluna
• Compila regras declarativas em expressoes Spark
• Gera score e status por coluna
• Aplica tratamento em lote (inclui colapso de espacos internos em texto DSC_)
• Executa pipeline antes/depois com comparativo
• Detecta avisos (informativos, nao alteram o score):
  • cardinalidade IDT_ fora do padrao;
  • similaridade categorica DSC_/TPO_ — mesma categoria com grafia diferente, por pontuacao/espacos (ex.: "S.A" vs "S/A");
  • nulos padronizados (N/D/0/1800-01-01) por coluna, marcando [DROP?] quando sao maioria.
• Registra em dq_historico.xlsx (abas Historico com % Colunas Aprovadas, Relatorio com Causa/Acao Sugerida, e Avisos)
• Suporta modo producao (valida o DataFrame ja tratado e marca situacao CORRIGIDO)

Requisitos

• Python 3.10+
• PySpark 3.4+

Instalacao

pip install -e .

Uso rapido

from pyspark.sql import SparkSession
from dq_engine import DataQualityEngine
from dq_engine.config.conventions import CONVENTIONS

spark = SparkSession.builder.getOrCreate()
engine = DataQualityEngine(spark=spark, conventions=CONVENTIONS)

df = spark.table("lakehouse.clientes")
result = engine.validate_table(df=df, table_name="CLIENTES", run_id="2026-07-02")

result.output_df.orderBy("column_name").show(truncate=False)

Dois modos de uso

Simulacao

df_tratado, result_before, result_after = engine.run_dq_pipeline(
    df=df,
    table_name="CLIENTES",
    output_dir="/mnt/dq_output",  # opcional: sem ele nada e gravado em arquivo
)

Producao

df_validado, result, _ = engine.run_dq_pipeline(
    df=df_ja_tratado,
    table_name="CLIENTES",
    output_dir="/mnt/dq_output",
    producao=True,
    ignore_columns_from_score=[],   # lista de colunas a tirar do score (veja abaixo)
)

Ignorar colunas no score (ignore_columns_from_score)

Para uma particularidade ja conhecida e nao corrigivel agora (ex.: IDT_X que na verdade deveria ser TPO_X), informe o nome exato da coluna:

df_validado, result, _ = engine.run_dq_pipeline(
    df=df_ja_tratado,
    table_name="CLIENTES",
    output_dir="/mnt/dq_output",
    producao=True,
    ignore_columns_from_score=["IDT_SITUACAO", "IDT_SUBCATEGORIA"],
)

• Vale igual em simulacao e producao: a coluna continua sendo validada, mas fica fora do quality_score/indicador da tabela (aba Historico) e do diagnostico de falha residual (aba Relatorio). Nao e manipular score, e priorizar.
• Em simulacao, a coluna tambem fica fora do trecho de ETL e nao e tratada na simulacao (o resultado simulado bate com o que o ETL gerado vai fazer); aparece numa linha "fora do plano (lista de excecao)".
• O estado real da coluna e mostrado no console/.txt (subsecao "Colunas monitoradas") e gravado na aba Avisos do Excel (Tipo Aviso = COLUNA_MONITORADA) para ajuste posterior. No console, a coluna monitorada e uma entrada so (os avisos de regra, IDT_ e grafia dela entram resumidos: avisos: idt_cardinality (5 valores)); no Excel, cada aviso continua num registro proprio, para o painel contar por regra e categoria.
• [] (ou omitir) = todas as colunas entram no score.

Via main.py: python main.py --table ... --producao --ignore-columns-producao IDT_X (flag repetivel), ou no .env: DQ_PRODUCAO=true e DQ_IGNORE_COLUMNS_PRODUCAO=IDT_X,IDT_Y.

Como ler o output

O bloco AVISOS abre com o RESUMO POR COLUNA: uma linha por coluna com todos os avisos dela, por exemplo:

  RESUMO POR COLUNA — 3 coluna(s) com aviso
  DSC_PLACA     placa padrao antigo (2); nulo fora do padrao (1)
  NUM_CHASSI    formato invalido (1); DV de chassi divergente (1)
  DSC_OUTRA     FALHA RESIDUAL score 0.0000 (upper)

Com output_dir, o console mostra so o resumo; o detalhe por regra (exemplos e acao) fica no .txt salvo. Sem output_dir, o console mostra o resumo e o detalhe. O detalhe tem dois blocos:

1. PONTO DE ATENCAO — nao reduz o score. Um bloco unico, em subsecoes:
   • IDT_ fora do padrao de dominio (mais de 2 valores distintos alem de N/D: provavel erro de prefixo, deveria ser TPO_/DSC_);
   • regras WARNING (ex.: ano suspeito em datas, campos estritos como CPF/CNPJ/placa/chassi), sempre com exemplos dos valores que dispararam a regra. Colunas DSC_ (campo aberto) nao validam encoding_corruption;
   • categorias duplicadas por grafia (S.A x S/A);
   • colunas monitoradas (fora do score).
2. FALHA RESIDUAL — reduz o score. Colunas que seguem com falha. Cada regra aparece separada, com o que significa, quantas linhas, exemplos daquela regra e a acao:
   • "o tratamento corrige" (nulo fora do padrao, placeholder, maiusculas...): se aparece em producao, a tabela nao foi reprocessada com o data_clean atual;
   • "o tratamento NAO corrige" (texto so com numero, encoding, dominio, formato): corrigir na fonte ou rever o prefixo da coluna.

Tudo isso vai tambem para o Excel (aba Avisos), com coluna, regra, categoria, quantidade e mensagem (incluindo exemplos).

Regras de tratamento de texto

Todo tratamento de string converte para MAIUSCULAS, remove acentos e trata nulos (N/D), exceto:

• COD_ — mantem o texto como veio (so trim): sem upper, preserva pontuacao, acentos e minusculas — codigos/chaves nao mudam. O nulo de COD_ e sempre "0" (nunca N/D), inclusive em COD_ com regra propria pelo nome (COD_CHASSI, COD_UF, COD_CPF...), que mantem a validacao/avisos do campo mas devolvem "0" no nulo;
• colunas com PATH, LINK ou ANEXO no nome (qualquer prefixo) e DSC_ com BUCKET/CAMINHO (treat_dsc_path) — nao passam para maiusculas: preservam o caminho/link e usam minusculas.

Datas (DAT_/DTH_) mantem o tipo: coluna date continua date e coluna timestamp continua timestamp (verificacao pelo schema). So coluna texto (ou inteiro, ex.: 20240101) e classificada pelas amostras: com hora diferente de 00:00:00 e timestamp, senao date. O trecho de ETL sugere e explica a escolha:

      "DAT_NASCIMENTO": treat_date("DAT_NASCIMENTO", tipo="date"),
      "DAT_ABERTURA":   treat_datetime("DAT_ABERTURA"),
  # datas: schema date/timestamp mantém o tipo; texto pelas amostras (com hora = timestamp)
  #   DAT_ABERTURA: timestamp (schema)
  #   DAT_NASCIMENTO: date (schema)

Chamada direta sem tipo: treat_date e treat_datetime devolvem timestamp (compativel com os ETLs em producao). Formatos aceitos: ISO, dd/MM/yyyy, dd/MM/yyyy HH:mm[:ss] e yyyyMMdd; o parse so e tentado no formato que casa com o texto, entao funciona tambem com spark.sql.ansi.enabled=true. Nulo/invalido vira 1800-01-01 00:00:00, assim como datas com ano abaixo de suspicious_year_min (default 1900, ex.: 0001-01-01). Datas futuras sao mantidas e so geram aviso (ano suspeito). Tabelas REP_/REF_ (colunas date) nao tem tratamento especifico no dq-engine.

treat_string (NOM_, TPO_, END_...): trim, maiusculas, sem acento, espacos colapsados e nulo disfarcado (VAZIO, N/A, TESTE, 0, vazio, texto sem letra/numero) -> N/D. Hifen entre letras vira espaco (ANA-MARIA -> ANA MARIA). Ponto e os demais caracteres especiais ficam (SIMONE.SILVA, M&M). Para tirar, use treat_string("X", limpar_especiais=True): simbolo vira espaco (SIMONE SILVA, M M). 999, 9999 e -9999 nunca viram N/D.

DSC_ (texto livre ou categorica, de baixa cardinalidade) e STR_ usam a mesma funcao, treat_dsc. DSC_ capturadas por nome (ex.: DSC_PLACA) mantem o tratamento especifico mesmo com baixa cardinalidade.

• Simbolos em qualquer posicao: treat_dsc mantem letras, numeros, espaco e a pontuacao . , ; : - / ( ) [ ] ' " ? ! % & + = @ _; o resto (* # | ~ ^ $ < >, emojis...) e removido tambem no meio do texto ("EVENTO** RELATO" -> EVENTO RELATO).
• Ruido nas bordas: simbolos soltos no inicio/fim saem ("SUZANO *" -> SUZANO, "SUZANO ." -> SUZANO), mas (/[ no inicio e )/]/% no fim ficam ("... (GASOLINA / FLEX)" mantem o ) final).
• Placeholders: X, TESTE, sequencia de zeros (0, 000000) e o mesmo digito repetido 3+ vezes (111, 11111111111111111111111) viram N/D. 1, 22, 208 continuam valor.
• Ponto final: . colado em letra/numero fica (abreviacao: 16V AUT.); ponto solto (SUZANO .) ou repetido (AUT..) sai.
• Localidade (CIDADE, MUNICIPIO, ESTADO, BAIRRO, PAIS, LOCALIDADE, DISTRITO, REGIAO como palavra inteira do nome — DSC_ESTADO sim, DSC_PRESTADOR nao — via treat_dsc ou treat_string): limpeza estrita, so letras, numeros, espaco, hifen e apostrofo (UBERABA. -> UBERABA, Embu-Guaçu -> EMBU-GUACU, Santa Bárbara D'Oeste -> SANTA BARBARA D'OESTE). UF usa treat_uf (so sigla). Cidade com UF junto (VILA VELHA, ES, UBERABA - MG, CAMPINAS (SP)) gera aviso cidade_com_uf sugerindo separar em duas colunas (cidade e UF). VAL_/QTD_/NUM_/COD_ com CIDADE no nome (ex.: VAL_CUSTO_CIDADE_MENSAL) seguem a regra do prefixo.
• Texto so com numero: em DSC aberta (alta cardinalidade) e so aviso, nao reduz o score; em DSC categorica (baixa cardinalidade) continua erro, porque numero no meio das categorias costuma ser codigo no lugar da descricao. Colunas com MODELO no nome nao usam a regra (208, 3008, 500 sao modelos reais).

Funcoes treat_* do trecho de ETL sao sempre as mesmas desde a 0.7.2 — ajustes novos mudam o comportamento dessas funcoes, sem criar nomes novos. Para os ajustes valerem nos ETLs que ja estao em producao (que chamam a funcao gravada no trecho da epoca), cada treat_* aplica a regra da familia pelo nome da coluna:

COD_ com campo estrito (COD_CHASSI, COD_UF, COD_CPF...) continua com a validacao do campo, mas o nulo e "0".

Tipos numericos

Excecoes pelo nome continuam com a regra propria: NUM_APOLICE/NUM_DOCUMENTO (texto), NUM_TELEFONE, NUM_CHASSI, NUM_CPF.../NUM_CNPJ.... Atencao: int vai ate 2.147.483.647 — colunas NUM_ com mais digitos (protocolos, notas fiscais) estouram.

QTD_, VAL_ e PERC_ seguem sempre o prefixo, mesmo com UF/CPF/PLACA no nome (QTD_ORDEM_UF_MES e int, nao UF). A sigla UF so casa como palavra inteira do nome (COD_UF, UF_ORIGEM).

Sentinelas numericos (9999, -999, -9999 — numeric_sentinel_values — e 999 so em VAL_), comuns em fill_na de ETLs de score (VAL_SCORE_* = 999, VAL_FRANQUIA = -9999.0), geram aviso numeric_sentinel (categoria NULO, nao afeta o score). O valor nao e trocado no tratamento; o ajuste e no fill_na do ETL (nulo numerico padrao 0).

999 e 998 sao valores reais, nunca nulo: COD_SEGURADORA 999 = loja propria / TODAS CIAS, 998 = RLP/RPS, NUM_SEQUENCIA 999 = loja propria. treat_cod/treat_int mantem o valor e a validacao nao acusa nulo fora do padrao.

Campos estritos: so aviso, valor invalido mantido

CPF, CNPJ, CEP, UF, telefone, placa e chassi (resolvidos pelo nome da coluna) nao afetam o score — todas as regras sao WARNING e aparecem na aba Avisos. No tratamento:

• nulo disfarcado (VAZIO, 0, TESTE, 000.000.000-00, vazio) -> N/D;
• valor valido -> forma padronizada (sem mascara; telefone como (XX) XXXXX-XXXX);
• valor invalido -> mantido como veio, com aviso "valor mantido; corrigir na origem" (ex.: CNPJ incompleto continua incompleto).

Avisos especificos, alem de formato/nulo:

No modo simulacao, problemas que o tratamento corrigiu (ex.: nulo disfarcado -> N/D) tambem continuam na aba Avisos, com a indicacao "trocado(s) por N/D no tratamento".

Colunas _NORMATIZADO (PKs: chassi, placa, CPF, CNPJ)

Para colunas de chassi, placa, CPF e CNPJ (normatizado_rule_sets), o trecho de ETL ganha um bloco extra que cria <COLUNA>_NORMATIZADO. Nessa coluna, valor valido vira a forma padronizada e valor invalido ou nulo vira N/D ("0" em COD_). A coluna original segue o tratamento normal (invalido mantido + aviso):

df = df.withColumns({
    "NUM_CHASSI_NORMATIZADO": treat_chassi("NUM_CHASSI", normatizado=True),
    "NUM_CPF_NORMATIZADO":    treat_cpf("NUM_CPF", normatizado=True),
})

normatizado=True e um parametro das mesmas treat_* (sem nome novo). NUM_APOLICE nao tem validacao de formato (e texto) e por isso nao ganha coluna _NORMATIZADO.

Onde os arquivos sao salvos

Quando output_dir e informado, o DqOutputSaver grava:

{output_dir}/
├── table_output/
│   └── {TABELA}_{TIMESTAMP}.txt        # relatorio detalhado de cada execucao
└── output_consolidado/
    └── dq_historico_v2.xlsx            # historico acumulado (nome via historico_filename)

Abas do Excel: Historico (indicador/score por execucao), Relatorio (diagnostico por coluna) e Avisos (todos os avisos acima).

No main.py a configuracao e:

Se o Excel estiver aberto, o save avisa e nao grava o registro: feche e rode de novo.

Leitura via Snowflake (opcional)

O engine sempre recebe um DataFrame Spark comum — a origem dos dados e desacoplada. Para descobrir e ler tabelas do Snowflake por schema, instale o extra snowflake:

pip install -e ".[snowflake]"

from dq_engine.sources.snowflake_source import SnowflakeReader, discover_tables

reader = SnowflakeReader()  # credenciais via variaveis de ambiente (.env)
table_names = discover_tables(reader, schema_like="%NOME_DO_SCHEMA%")
df = reader.read(spark, table_names[0])

result = engine.validate_table(df=df, table_name=table_names[0], run_id="2026-07-14")

O script main.py automatiza o fluxo (descobre tabelas do schema -> roda dq-engine -> salva consolidado/historico) e pode ser agendado no Sonata:

python main.py --schema-like "%NOME_DO_SCHEMA%"

O padrao LIKE tambem pode vir do .env (variavel TABLE_SCHEMA), dispensando --schema-like em toda chamada.

Tambem e possivel rodar uma lista fixa de tabelas em vez de (ou combinada com) o schema, via --table (repetivel) ou pela variavel TABLE_LIST no .env (nomes completos DATABASE.SCHEMA.TABELA separados por virgula):

python main.py --table TRUSTED.MAX_APOLICE.TABELA_1 --table TRUSTED.MAX_APOLICE.TABELA_2

Tambem e possivel apontar um YAML com lista simples de tabelas (como to_process.yaml) usando --tables-yaml ou a variavel TABLE_LIST_FILE no .env:

python main.py --tables-yaml to_process.yaml

Sem --schema-like/TABLE_SCHEMA, o main.py roda so as tabelas informadas (--table, TABLE_LIST ou --tables-yaml), sem descoberta via INFORMATION_SCHEMA; com os dois informados, a lista filtra o resultado da descoberta.

Tabelas processadas com sucesso ficam registradas em processed_tables.yaml (arquivo temporario, desacoplado do dq-engine) e sao puladas nas proximas execucoes; use --rerun-processed para forcar tudo de novo.

API principal

• DataQualityEngine.validate_table
• DataQualityEngine.apply_treatment
• DataQualityEngine.run_dq_pipeline
• DataQualityEngine.get_column_treatment_contexts
• DataQualityResult.table_summary
• DataQualityResult.failed_columns

Configuracao

Arquivo central de convencoes:

• src/dq_engine/config/conventions.py

Documentacao

• docs/architecture.md
• docs/rules.md