Validação de Schema e Frameworks de Qualidade de Dados: Construindo Pipelines Robustos

Validação de Schema e Frameworks de Qualidade de Dados: Construindo Pipelines Robustos

Em pipelines ETL complexos, a qualidade dos dados é tão crítica quanto a quantidade. Enquanto Pandas oferece ferramentas básicas de limpeza, frameworks especializados como Great Expectations e Pydantic fornecem camadas de validação declarativas que funcionam como contratos de dados, garantindo conformidade em múltiplos estágios do pipeline.

O Problema Fundamental: Validação Reativa vs. Proativa

Tradicionalmente, validação de dados ocorre após o carregamento—você descobre problemas quando o dashboard quebra ou análises produzem resultados suspeitos. Frameworks modernos invertem essa lógica: definem expectativas antes do processamento, capturando anomalias na origem.

A diferença é profunda. Com Pandas puro, você valida assim:

df = pd.read_csv('dados.csv')
if df['idade'].min() < 0:
    print('Erro: idades negativas detectadas')
    # Agora você já processou dados inválidos

Com Great Expectations, você declara expectativas como cidadão de primeira classe:

from great_expectations.dataset import PandasDataset

df = PandasDataset(pd.read_csv('dados.csv'))
df.expect_column_values_to_be_between('idade', min_value=0, max_value=150)
df.expect_column_values_to_not_be_null('email')
validation_result = df.validate()

Great Expectations: Validação Declarativa em Escala

Great Expectations separa definição de expectativas (o quê validar) de execução (quando e como). Isso permite reutilização e versionamento de contratos de dados.

Internamente, Great Expectations mantém um registro de todas as validações executadas, gerando relatórios HTML que documentam a linhagem de qualidade. Para dados em Pandas:

import great_expectations as ge

context = ge.get_context()
suite = context.create_expectation_suite('meu_suite')

df = ge.read_pandas(pd.read_csv('vendas.csv'))
df.expect_column_to_exist('data_venda')
df.expect_column_values_to_match_regex('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')
df.expect_table_row_count_to_be_between(min_value=1000, max_value=1000000)

validation_result = df.validate(expectation_suite=suite)
print(validation_result['success'])  # True/False

Um detalhe crítico: Great Expectations não falha por padrão—retorna um objeto de resultado que você inspeciona. Isso permite lógica condicional sofisticada: talvez 5% de nulos sejam aceitáveis, mas 50% não.

Pydantic: Validação Estruturada em Tempo de Execução

Enquanto Great Expectations é orientado a datasets, Pydantic valida estruturas individuais. É ideal para validar registros antes de inseri-los em DataFrames ou APIs.

from pydantic import BaseModel, Field, validator
from typing import Optional

class Venda(BaseModel):
    id: int
    valor: float = Field(gt=0, le=1000000)
    email: str
    data: str
    
    @validator('email')
    def email_valido(cls, v):
        if '@' not in v:
            raise ValueError('Email inválido')
        return v.lower()

# Uso em pipeline
registros = [{'id': 1, 'valor': 99.99, 'email': 'USER@EXAMPLE.COM', 'data': '2024-01-01'}]
vendas_validadas = [Venda(**r) for r in registros]

Pydantic é estrito por padrão—falha na primeira violação. Great Expectations é permissivo—coleta todas as violações. Escolha baseado em seu padrão de falha desejado.

Trade-offs e Considerações de Performance

Great Expectations adiciona overhead significativo para datasets pequenos (<10MB), mas escala bem para dados maiores porque pode usar backends como Spark. Pydantic é leve mas não paraleliza validação de forma nativa.

Para pipelines com milhões de registros, considere validação em camadas: Pydantic na ingestão (rápido, falha cedo) e Great Expectations em checkpoints críticos (completo, documentado).

Um edge case importante: ambos frameworks podem mascarar problemas de lógica de negócio. Uma coluna pode ter valores válidos sintaticamente mas semanticamente incorretos. Combine validação estrutural com regras de negócio customizadas.

Integração com Pipelines Pandas

Para máxima compatibilidade com seu stack atual:

def validar_e_limpar(df: pd.DataFrame) -> tuple[pd.DataFrame, dict]:
    df_validado = ge.read_pandas(df)
    df_validado.expect_column_values_to_not_be_null('chave_primaria')
    resultado = df_validado.validate()
    
    if not resultado['success']:
        falhas = resultado['results']
        df_limpo = df.dropna(subset=['chave_primaria'])
        return df_limpo, {'validacoes_falhadas': len(falhas)}
    
    return df, {'validacoes_falhadas': 0}

Isso permite integração gradual: comece com validações simples, evolua para suites complexas conforme sua confiança aumenta.