Engenharia de Dados Defensiva: Implementação de referência para validação de dados em tempo de execução (Runtime Validation), substituindo contratos estáticos por código executável que bloqueia dados corrompidos na ingestão.
Na stack de dados moderna, a maioria das validações ocorre tarde demais (geralmente via dbt test após a carga no Data Warehouse - Late Binding). Isso cria o fenômeno dos "Pipelines Zumbis": fluxos de ETL que executam com sucesso técnico (retornam código 0), mas trafegam dados de baixa qualidade, gerando:
- Custo de Nuvem Desperdiçado: Processar e armazenar lixo custa dinheiro (Compute/Storage). A regra "1-10-100" dita que limpar dados no Warehouse custa 10x mais que na fonte.
- Data Downtime: Dashboards quebram silenciosamente devido a mudanças não detectadas no upstream.
- Pobreza Estatística: Testes SQL padrão verificam nulidade e chaves, mas falham em detectar anomalias de distribuição (ex: desvios de média, outliers contextuais e schema drifts).
Este projeto implementa um Data Contract Executável usando a biblioteca Pandera. Diferente da validação orientada a objetos (Pydantic), o Pandera realiza validação vetorizada de alta performance em DataFrames.
O pipeline não falha cegamente (o que pararia o negócio). Ele atua como um disjuntor inteligente que separa os dados em dois fluxos:
- ✅ Fluxo Silver: Dados 100% aderentes ao contrato seguem para o Data Lake/Warehouse.
- 🚫 Fluxo Quarentena (DLQ): Linhas inválidas são isoladas em arquivos separados, enriquecidas com metadados de erro (JSON) para depuração rápida e reprocessamento futuro.
- Python 3.9+
- Pandera: Definição de esquema, validação estatística e Lazy Validation.
- Pandas: Manipulação de dados em memória.
- Apache Airflow (Simulado): Estrutura de tasks para orquestração e alertas.
Clone o repositório e instale as dependências (versões travadas para reprodutibilidade):
git clone https://github.com/seu-usuario/pandera-circuit-breaker.git
cd pandera-circuit-breaker
pip install -r requirements.txtExecute o script utilitário para criar um arquivo CSV contendo propositalmente dados válidos e inválidos (IDs duplicados, erros de regex, anomalias de valor e moeda incorreta):
python setup_data.py
# Output: 📁 Massa de teste gerada em: data/raw_transactions.csvRode o processo de ingestão. O script lerá o CSV, validará contra o contrato e dividirá o fluxo automaticamente.
python -m src.pipelineVerifique a pasta output/ para ver a separação física dos dados:
- Dados Limpos:
output/silver/transactions_clean.csv - Dados Rejeitados:
output/quarantine/transactions_quarantine.csv- Dica: Abra este arquivo e verifique a coluna
dq_error_log. Ela contém um JSON detalhando exatamente qual regra foi violada (ex:check_anomalia_zscore,str_matches).
- Dica: Abra este arquivo e verifique a coluna
O coração do projeto é o SchemaModel definido em src/contracts.py. Note o uso de validação estatística (Z-Score) impossível de fazer facilmente em SQL:
class ContratoTransacoes(pa.DataFrameModel):
# 1. Validação de Sintaxe (Regex)
# Garante IDs corporativos no padrão correto
customer_id: Series[str] = pa.Field(str_matches=r'^[A-Z]{3}-\d{5}$')
# 2. Validação Semântica (Check Estatístico / Z-Score)
@pa.check("amount", name="check_anomalia_zscore")
def valida_anomalia(cls, series):
# Falha se o desvio padrão for > 3 (anomalia estatística)
# Protege contra erros de conversão de moeda sistêmicos
return abs((series - series.mean()) / series.std()) < 3
class Config:
strict = True # Bloqueia colunas não mapeadas (Schema Drift)
coerce = True # Corrige tipos automaticamente se possívelSinta-se à vontade para abrir Issues ou PRs. O objetivo é criar o padrão-ouro de validação.
MIT