Validação e Qualidade de Dados

Verificações de esquema e estatísticas nas fronteiras do pipeline capturam ingestões ruins antes que poluam os marts - Pandera para DataFrames Python, Great Expectations para contratos de dados documentados.

Receita

Cartão de receita de referência rápida - pronto para copiar e colar.

import pandera.pandas as pa
 
schema = pa.DataFrameSchema({
    "order_id": pa.Column(int, unique=True, nullable=False),
    "revenue": pa.Column(float, pa.Check.ge(0)),
    "region": pa.Column(str, pa.Check.isin(["East", "West", "Central"])),
})
 
def validate_orders(df):
    return schema.validate(df, lazy=True)

Quando usar isso:

• Cada estágio ETL antes de sobrescrever escritas
• Testes de CI em amostras Parquet de fixture
• Dashboards de SLA sobre frescor e volume
• Dados regulatórios ou financeiros que necessitam de trilha de auditoria

Exemplo de Trabalho

import pandas as pd
import pandera.pandas as pa
from pandera import Check, Column, DataFrameSchema
 
orders_schema = DataFrameSchema(
    {
        "order_id": Column(int, unique=True, nullable=False),
        "revenue": Column(float, checks=Check.ge(0)),
        "region": Column(str, checks=Check.isin(["East", "West", "Central"])),
        "ordered_at": Column("datetime64[ns, UTC]", nullable=False),
    },
    strict=False,
    coerce=True,
)
 
def load_orders(path: str) -> pd.DataFrame:
    df = pd.read_parquet(path)
    try:
        return orders_schema.validate(df, lazy=True)
    except pa.errors.SchemaErrors as err:
        # Quarentena de linhas ruins para inspeção
        failure_cases = err.failure_cases
        failure_cases.to_csv("quarantine/schema_failures.csv", index=False)
        raise
 
# Verificação estilo Great Expectations (API GE conceitual)
def gx_style_checks(df: pd.DataFrame) -> None:
    assert len(df) > 0, "lote vazio"
    assert df["order_id"].is_unique.all()
    assert df["revenue"].between(0, 1_000_000).mean() > 0.99

O que isso demonstra:

• Esquema Pandera com coerção e relatório de múltiplos erros lazy
• Artefato de quarentena em caso de falha
• Asserções leves complementando esquemas formais
• Tipagem de coluna datetime UTC

Mergulho Profundo

Como Funciona

• Pandera valida a forma do DataFrame, tipos de dados e verificações personalizadas em Python.
• Great Expectations armazena Expectation Suites, valida lotes, renderiza Data Docs.
• Falhar fechado: a tarefa orquestradora não deve promover o mart em caso de erro de validação.
• Combine esquema (estrutura) com verificações de anomalias (volume, taxa de nulos).

Tipos de Verificação

Notas de Python

# Polars com pandera.polars
import polars as pl
import pandera.polars as pa
 
schema = pa.DataFrameSchema({"id": pa.Column(int)})
schema.validate(pl.DataFrame({"id": [1, 2]}))

Armadilhas

• Validando apenas em notebook - a produção ignora as verificações. Correção: chamar o validador em @task antes da escrita.
• lazy=False em esquemas amplos - para no primeiro erro. Correção: lazy=True para relatório completo de falhas.
• Contagens de linhas exatas frágeis - instáveis com tráfego real. Correção: verificações de intervalo vs. média móvel.
• Falhas apenas de aviso - dados ruins ainda são enviados. Correção: levantar exceção e bloquear sensores downstream.
• Evolução de esquema não versionada - implantação falha silenciosamente. Correção: marcar versões de esquema com git tag junto com o código do pipeline.

Alternativas

FAQs

import pytest
import pandera.errors
 
def test_rejects_negative_revenue(bad_df):
    with pytest.raises(pandera.errors.SchemaError):
        orders_schema.validate(bad_df)

Relacionado

• Limpeza e Transformação de Dados - correções antes de validar
• dbt com Python - testes de warehouse
• Boas Práticas de Engenharia de Dados - portões de qualidade
• Confiabilidade de Fluxo de Trabalho - falhar vs. tentar novamente

Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.