Frameworks de Qualidade de Dados

Padrões de produção para implementar qualidade de dados com Great Expectations, testes dbt e contratos de dados para garantir pipelines de dados confiáveis.

Quando Usar Esta Habilidade

• Implementar verificações de qualidade de dados em pipelines
• Configurar validação com Great Expectations
• Construir suítes de testes dbt abrangentes
• Estabelecer contratos de dados entre equipes
• Monitorar métricas de qualidade de dados
• Automatizar validação de dados em CI/CD

Conceitos Principais

1. Dimensões de Qualidade de Dados

2. Pirâmide de Testes para Dados

          /\
         /  \     Integration Tests (cross-table)
        /────\
       /      \   Unit Tests (single column)
      /────────\
     /          \ Schema Tests (structure)
    /────────────\

Início Rápido

Configuração do Great Expectations

# Install
pip install great_expectations

# Initialize project
great_expectations init

# Create datasource
great_expectations datasource new

# great_expectations/checkpoints/daily_validation.yml
import great_expectations as gx

# Create context
context = gx.get_context()

# Create expectation suite
suite = context.add_expectation_suite("orders_suite")

# Add expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)

# Validate
results = context.run_checkpoint(checkpoint_name="daily_orders")

Padrões detalhados e exemplos práticos

A documentação detalhada dos padrões está em references/details.md. Leia este arquivo quando a navegação de nível superior for insuficiente.

Resumo: {total_passed}/{total_tables} tabelas passaram")

    report.append("")

    for table, result in results.items():
        status = "✅" if result.passed else "❌"
        report.append(f"### {status} {table}")
        report.append(f"- Expectativas: {result.total_expectations}")
        report.append(f"- Falharam: {result.failed_expectations}")

        if not result.passed:
            report.append("- Verificações falhas:")
            for detail in result.details:
                if not detail["success"]:
                    report.append(f"  - {detail['expectation']}: {detail['observed_value']}")
        report.append("")

    return "\n".join(report)

Uso

context = gx.get_context() pipeline = DataQualityPipeline(context)

tabelas_para_validar = { "pedidos": "suite_pedidos", "clientes": "suite_clientes", "produtos": "suite_produtos", }

resultados = pipeline.run_all(tabelas_para_validar) relatorio = pipeline.generate_report(resultados)

Falhar pipeline se alguma tabela falhou

if not all(r.passed for r in resultados.values()): print(relatorio) raise ValueError("Verificações de qualidade de dados falharam!")

## Melhores Práticas

### O Que Fazer

- **Testar cedo** - Validar dados de origem antes das transformações
- **Testar incrementalmente** - Adicionar testes à medida que encontra problemas
- **Documentar expectativas** - Descrições claras para cada teste
- **Alertar sobre falhas** - Integrar com monitoramento
- **Versionar contratos** - Rastrear alterações de esquema

### O Que Não Fazer

- **Não testar tudo** - Focar em colunas críticas
- **Não ignorar avisos** - Eles geralmente precedem falhas
- **Não pular a atualização** - Dados desatualizados são dados ruins
- **Não codificar limites fixos** - Usar linhas de base dinâmicas
- **Não testar isoladamente** - Testar relacionamentos também