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"],
)

• So tem efeito com producao=True (em simulacao e ignorado).
• A coluna continua sendo validada; apenas fica fora do quality_score/indicador da tabela e do diagnostico de falha residual. Nao e manipular score, e priorizar.
• 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.
• [] (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 output tem dois blocos de aviso:

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 apos o tratamento: corrigir na fonte.

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_ — so upper + trim, preservando pontuacao (nulo vira "0");
• colunas DSC_ com PATH/BUCKET/CAMINHO (treat_dsc_path) — preservam o caminho e usam minusculas.

Datas (DAT_/DTH_) sao sempre timestamp (padrao das tabelas trusted TRU_): treat_date e treat_datetime devolvem TimestampType, preservam a hora e, para data sem hora, usam 00:00:00. Nulo/invalido vira 1800-01-01 00:00:00. Tabelas REP_/REF_ (colunas date) nao tem tratamento especifico no dq-engine.

DSC_ de baixa cardinalidade (<= dsc_cardinality_threshold) sao reclassificadas como dsc_category e tratadas com treat_dsc_category; as demais usam treat_dsc (preserva espacos e pontuacao interna). DSC_ capturadas por nome (ex.: DSC_PLACA) mantem o tratamento especifico mesmo com baixa cardinalidade.

• Placeholders em DSC_: X, TESTE e sequencia de zeros (0, 000000) viram N/D.
• Ruido nas bordas de DSC_: treat_dsc remove simbolos soltos no inicio/fim ("SUZANO *" -> SUZANO), mas preserva parenteses/colchetes/chaves ("... (GASOLINA / FLEX)" mantem o ) final).
• STR_ usa treat_str (igual a treat_dsc, sem a troca de placeholders de DSC_).

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.

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".

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), Avisos (todos os avisos acima) e Distribuicao (opcional, include_distribution=True).

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