Qualidade de Dados - Soda Core¶

Esta página documenta os padrões e implementação de Data Quality utilizados em todos os projetos de dados da Lavoro, com foco na ferramenta Soda Core para validação automatizada.

🎯 Visão Geral¶

O que é Data Quality?¶

Data Quality (Qualidade de Dados) é o processo de garantir que os dados estejam corretos, completos, consistentes e confiáveis para uso em análises e tomadas de decisão.

Por que Data Quality é crítico?¶

Dados de baixa qualidade podem levar a:

❌ Decisões empresariais incorretas
❌ Perda de confiança nos dashboards e relatórios
❌ Retrabalho e aumento de custos operacionais
❌ Impactos negativos em regulamentações e compliance
❌ Problemas de integração entre sistemas

Princípios de Qualidade na Lavoro¶

Nossa estratégia de qualidade de dados é baseada em 6 dimensões fundamentais:

Dimensão	Descrição	Exemplo
Completude	Todos os dados obrigatórios estão presentes	Não ter valores nulos em `pedido_id`
Unicidade	Não há registros duplicados onde não deveria haver	IDs únicos em dimensões
Validade	Dados seguem regras e formatos esperados	E-mails com formato válido, datas válidas
Consistência	Dados são uniformes entre sistemas e tabelas	Mesmo cliente tem mesmo ID em todas as tabelas
Precisão	Dados refletem a realidade corretamente	Saldo de estoque corresponde ao físico
Atualidade (Freshness)	Dados estão atualizados conforme SLA definido	Faturamento atualizado diariamente

🛠️ Por que Soda Core?¶

Soda Core é uma ferramenta open-source para validação de qualidade de dados que oferece:

Principais Vantagens¶

✅ Configuração declarativa: Validações em YAML simples e legível
✅ Integração nativa com Airflow: Executa automaticamente nos pipelines
✅ Suporte a BigQuery: Conexão direta e otimizada
✅ Checks dinâmicos: Templates reutilizáveis com Jinja2
✅ Sampling inteligente: Mostra exemplos de falhas para debugging
✅ Métricas detalhadas: Estatísticas e logs estruturados

Comparação com Alternativas¶

Característica	Soda Core	Great Expectations	dbt Tests
Curva de aprendizado	✅ Baixa (YAML)	⚠️ Média (Python)	✅ Baixa (YAML)
Integração Airflow	✅ Nativa	⚠️ Customizada	⚠️ Via dbt operator
BigQuery performance	✅ Otimizado	⚠️ Genérico	✅ Otimizado
Checks dinâmicos	✅ Templates Jinja	⚠️ Python	⚠️ Limitado
Custo	✅ Free (Core)	✅ Free	✅ Free

🏗️ Arquitetura¶

Integração com Pipelines Airflow¶

O Soda Core está totalmente integrado aos pipelines de dados através de um operador customizado que valida dados antes de carregá-los nas tabelas finais:

graph TB
    A[Airflow DAG] --> B[BQInsertJobOperator]
    B --> C[1. Criar Tabela Temporária]
    C --> D[2. SodaValidator]
    D --> E[3. Executar Checks]
    E --> F[3.1. Checks Dinâmicos]
    E --> G[3.2. Checks de Colunas]
    E --> H[3.3. Checks Genéricos]
    E --> I[3.4. Checks Customizados]
    F --> J{4. Validar Resultados}
    G --> J
    H --> J
    I --> J
    J -->|✅ OK| K[5. Carregar Tabela Final]
    J -->|❌ Falha| L[5. Registrar Falhas]
    K --> M[6. Remover Tabela Temp]
    L --> M

    style D fill:#4285f4,color:#fff
    style J fill:#fbbc04,color:#333
    style K fill:#34a853,color:#fff
    style L fill:#ea4335,color:#fff

Fluxo de Validação Detalhado¶

1. Criação de Tabela Temporária

-- Cria tabela temporária para validação
CREATE TABLE `projeto.dataset.tabela_temp_20250115` AS
SELECT * FROM (
    -- Query original do pipeline
)

2. Detecção Automática de Particionamento

O sistema detecta automaticamente se a tabela é particionada e aplica filtros para otimizar performance:

# Detecta coluna de partição
partition_info = get_partition_info('fato_faturamento')
# Resultado: {'is_partitioned': True, 'column': 'data_emissao'}

# Aplica filtro automaticamente em todas as validações
WHERE DATE(data_emissao) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)

3. Execução de Validações em Camadas

As validações são executadas em 4 camadas sequenciais:

Layer 1: Validações Dinâmicas (column_mappings.yml)
  ├─ Validação de emails
  ├─ Validação de CPF/CNPJ
  └─ Validação de freshness

Layer 2: Validações de Missing Data (auto-geradas)
  └─ % de valores ausentes para cada coluna

Layer 3: Validações Genéticas (generic_*_checks.yml)
  ├─ Diferença de contagem de registros
  └─ Detecção de duplicatas

Layer 4: Validações Customizadas ({tabela}_checks.yml)
  └─ Regras específicas de negócio

4. Logging Estruturado

Todas as falhas são registradas no BigQuery para análise:

-- Tabela de logs: lavoro_logs.soda_logs
SELECT
  created_at,
  project_id,
  dataset_id,
  table_id,
  message,
  categoria  -- 'Alerta' ou 'Erro'
FROM `driven-strength-374020.lavoro_logs.soda_logs`
WHERE DATE(created_at) = CURRENT_DATE()
ORDER BY created_at DESC;

5. Sampling Inteligente

Quando uma validação falha, o Soda automaticamente mostra amostras dos registros problemáticos nos logs do Airflow:

================================================================
🔍 Amostras de falha: E-mails com formato inválido
----------------------------------------------------------------
| email                | cliente_nome      | data_cadastro |
|----------------------|-------------------|---------------|
| joao.com             | João Silva        | 2025-01-15    |
| maria@               | Maria Santos      | 2025-01-14    |
| pedro@example        | Pedro Oliveira    | 2025-01-13    |
----------------------------------------------------------------
📊 Total: 3 amostras
================================================================

✅ Tipos de Validação¶

1. Validações Dinâmicas (column_mappings.yml)¶

Validações baseadas em templates reutilizáveis que podem ser aplicadas a múltiplas tabelas.

📧 Validação de E-mail¶

Objetivo: Garantir que campos de e-mail tenham formato válido e percentual aceitável de valores ausentes.

Configuração:

# dags/soda/column_mappings.yml
email_columns:
  template: |
    # Email validation
    - missing_percent({{ column }}):
        name: "🚨 E-mails ausentes acima de {{ missing_threshold }}%"
        warn: when >= {{ missing_threshold }}%
        fail: when > {{ missing_threshold | int * 2 }}%

    - invalid_count({{ column }}):
        valid format: email
        name: "⚠️ Formato inválido de e-mails"
        warn: when > {{ invalid_allowed }}
        fail: when > {{ invalid_allowed | int * 10 }}

  tables:
    referencia_elevor_cliente:
      column: email
      missing_threshold: 50    # Tolerância maior
      invalid_allowed: 10

Como Funciona:

Template Jinja2 é renderizado para cada tabela configurada
Variáveis {{ column }}, {{ missing_threshold }} são substituídas
Check é executado na tabela temporária
Resultados são categorizados em "warn" ou "fail"

Exemplo de Saída:

✅ PASS: E-mails ausentes: 8.5% (abaixo de 15%)
⚠️ WARN: Formato inválido de e-mails: 12 registros (acima de 10)

📅 Validação de Freshness (Atualidade)¶

Objetivo: Garantir que dados estão sendo atualizados dentro do prazo esperado.

Configuração:

# dags/soda/column_mappings.yml
date_columns:
  template: |
    - {{ rule }}:
        name: "{{ name }}"
        warn: when fail ratio > 0 and freshness_in_days <= {{ days * 5 }}
        fail: when freshness_in_days > {{ days * 5 }}

  tables:
    america_siagri_produto:
      checks:
        - name: "📅 Freshness inferior a 1 dia"
          rule: "freshness(data_cadastro) < 1d"
          vars:
            days: 1

    coram_agrotis_notas:
      checks:
        - name: "📅 Freshness inferior a 2 dias"
          rule: "freshness(data_emissao) < 2d"
          vars:
            days: 2

Como Funciona:

-- Validação gerada automaticamente
SELECT
  MAX(data_cadastro) as ultima_atualizacao,
  DATE_DIFF(CURRENT_DATE(), MAX(data_cadastro), DAY) as dias_desatualizacao
FROM `projeto.dataset.tabela_temp`
HAVING dias_desatualizacao > 1  -- Fail se > 1 dia

🆔 Validação de CPF/CNPJ (Exemplo Customizável)¶

Objetivo: Validar formato de documentos brasileiros.

Configuração:

# dags/soda/column_mappings.yml
cpf_cnpj_columns:
  template: |
    - invalid_cpf_cnpj_count:
        invalid_cpf_cnpj_count query: |
          SELECT COUNT(*) as invalid_count
          FROM {{ table_name }}
          WHERE LENGTH(REGEXP_REPLACE({{ column }}, r'[^0-9]', '')) NOT IN (11, 14)
            AND {{ column }} IS NOT NULL
        name: "🚨 CPF/CNPJ com tamanho inválido"
        fail: when > 0

  tables:
    dim_clientes:
      column: cpf_cnpj

2. Validações de Missing Data (Auto-geradas)¶

Objetivo: Detectar colunas com valores ausentes acima do esperado.

Como Funciona:

Para todas as colunas de todas as tabelas (exceto STRUCT e ARRAY), o sistema gera automaticamente uma validação de percentual de valores ausentes:

# Gerado automaticamente para cada coluna
checks for projeto.dataset.tabela_temp:
  - missing_pct_pedido_id:
      missing_pct_pedido_id query: |
        SELECT
          CASE WHEN COUNT(*) = 0 THEN 0
          ELSE (COUNT(*) - COUNT(pedido_id)) * 100.0 / COUNT(*)
          END AS missing_percent
        FROM `projeto.dataset.tabela_temp`
        WHERE DATE(data_processamento) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
      name: 'Coluna pedido_id com valores ausentes'
      warn: when = 100  # Alerta se 100% ausente

  - missing_pct_cliente_nome:
      missing_pct_cliente_nome query: |
        SELECT
          CASE WHEN COUNT(*) = 0 THEN 0
          ELSE (COUNT(*) - COUNT(cliente_nome)) * 100.0 / COUNT(*)
          END AS missing_percent
        FROM `projeto.dataset.tabela_temp`
        WHERE DATE(data_processamento) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
      name: 'Coluna cliente_nome com valores ausentes'
      warn: when = 100

Resultado:

⚠️ WARN: Coluna telefone_comercial com valores ausentes: 100%
✅ PASS: Coluna pedido_id com valores ausentes: 0%
✅ PASS: Coluna cliente_nome com valores ausentes: 2.3%

3. Validações Genéricas (generic_*_checks.yml)¶

Validações aplicadas a todas as tabelas automaticamente.

🔢 Diferença de Contagem de Registros¶

Objetivo: Detectar variações significativas entre execuções.

Configuração:

# dags/soda/generic_missing_data_checks.yml
checks for ${SODA_TABLE_NAME}:
  - diferenca_quantidade_tabelas:
      diferenca_quantidade_tabelas sql_file: "sql/check_diferenca_quantidade.sql"
      warn: when > 5 or < -5       # Alerta se diferença > 5%
      fail: when > 50 or < -50     # Falha se diferença > 50%

Query SQL:

-- dags/soda/sql/check_diferenca_quantidade.sql
WITH original AS (
  SELECT COUNT(*) AS valor_original
  FROM `${SODA_PROJECT}.${SODA_DATASET}.${SODA_ORIGINAL_TABLE_NAME}`
  WHERE ${SODA_ORIGINAL_PARTITION_FILTER}
),
novo AS (
  SELECT COUNT(*) AS valor_novo
  FROM `${SODA_TABLE_NAME}`
  WHERE ${SODA_TEMP_PARTITION_FILTER}
)
SELECT
  IFNULL(
    SAFE_DIVIDE((n.valor_novo - o.valor_original) * 100, o.valor_original),
    0
  ) AS pct_diferenca
FROM original o
CROSS JOIN novo n;

Exemplo de Resultado:

⚠️ WARN: Diferença de quantidade: +8.2% (1.250 → 1.352 registros)
❌ FAIL: Diferença de quantidade: -65% (5.000 → 1.750 registros)

🔄 Detecção de Duplicatas¶

Objetivo: Identificar registros duplicados considerando todas as colunas.

Configuração:

# dags/soda/generic_missing_data_checks.yml
checks for ${SODA_TABLE_NAME}:
  - linhas_duplicadas:
      name: "Validação de linhas duplicadas"
      linhas_duplicadas sql_file: "sql/check_linhas_duplicadas.sql"
      warn: when > 1

Query SQL:

-- dags/soda/sql/check_linhas_duplicadas.sql
WITH campos AS (
  SELECT ARRAY_AGG(column_name ORDER BY ordinal_position) AS colunas
  FROM `${SODA_PROJECT}.${SODA_DATASET}.INFORMATION_SCHEMA.COLUMNS`
  WHERE table_name = '${SODA_ORIGINAL_TABLE_NAME}'
),

linha AS (
  SELECT AS STRUCT *
  FROM `${SODA_TABLE_NAME}`
  WHERE ${SODA_TEMP_PARTITION_FILTER}
),

contagem AS (
  SELECT
    DISTINCT TO_JSON_STRING(linha) AS linha,
    CAST(COUNT(*) AS INT64) AS contagem_final
  FROM linha GROUP BY ALL
)

SELECT IFNULL(SUM(contagem_final), 0) AS linhas_duplicadas
FROM contagem
WHERE contagem_final > 1;

Como funciona: 1. Busca todas as colunas da tabela original 2. Converte cada linha completa em JSON para comparação 3. Conta linhas idênticas (considerando TODAS as colunas) 4. Retorna soma de linhas duplicadas

4. Validações Customizadas por Tabela¶

Objetivo: Implementar regras de negócio específicas para cada tabela.

Exemplo: dim_cliente_checks.yml

# dags/soda/dim_cliente_checks.yml
checks for ${SODA_TABLE_NAME}:
  # Campos obrigatórios não podem estar ausentes
  - missing_percent(id):
      name: "Porcentagem ids ausentes"
      fail: when > 0%

  - missing_percent(cnpj_cpf):
      name: "Porcentagem cnpj_cpf ausentes"
      fail: when > 0%

  # E-mails podem ter até 45% de ausência
  - missing_percent(email):
      name: "Porcentagem E-mails ausentes"
      warn: when > 0% and <= 45%
      fail: when > 45%

  # Formato de e-mail válido (até 15.000 inválidos permitidos)
  - invalid_count(email):
      name: "Quantidade E-mails inválidos"
      valid format: email
      fail: when > 15000

Exemplo: fato_faturamento_checks.yml (estrutura sugerida)

# dags/soda/fato_faturamento_checks.yml
checks for ${SODA_TABLE_NAME}:
  # Validações básicas de completude
  - row_count > 0

  - missing_percent(nota_fiscal_id):
      name: "IDs de nota fiscal ausentes"
      fail: when > 0%

  - missing_percent(data_emissao):
      name: "Datas de emissão ausentes"
      fail: when > 0%

  - missing_percent(valor_total):
      name: "Valores totais ausentes"
      warn: when > 5%
      fail: when > 10%

  # Validação de formato/valores
  - min(valor_total):
      name: "Valores totais negativos"
      fail: when < 0

  - max(valor_total):
      name: "Valores totais muito altos (outliers)"
      warn: when > 10000000  # Alerta se > 10 milhões

Nota: Para validações customizadas complexas com SQL, siga o padrão de dim_cliente_checks.yml e crie queries SQL separadas no diretório sql/.

📊 Tabelas Monitoradas¶

Cobertura por Projeto¶

CDP (Customer Data Platform)¶

Tabela	Validações	Checks Customizados
`referencia_elevor_cliente`	Email, Missing Data	❌

Lakehouse¶

Tabela	Validações	Checks Customizados
`dim_clientes`	Missing Data, Genéricas	✅
`dim_produtos`	Missing Data, Genéricas	❌
`fato_faturamento`	Missing Data, Genéricas	✅
`fato_estoque`	Missing Data, Genéricas, Freshness	✅

Obs: Todas as tabelas que utilizam BQInsertJobOperator recebem automaticamente validações de missing data e genéricas.

🔧 Como Adicionar Validações¶

Passo 1: Adicionar Validação Dinâmica¶

Caso de uso: Validar emails em uma nova tabela

# Editar: dags/soda/column_mappings.yml
email_columns:
  tables:
    # ... tabelas existentes ...

    # Nova tabela
    nova_geracao_sapb1_cliente:
      column: email_principal
      missing_threshold: 20     # 20% de ausência permitida
      invalid_allowed: 5        # 5 emails inválidos permitidos

Passo 2: Criar Checks Customizados¶

Caso de uso: Adicionar validações específicas para nova tabela

# Criar: dags/soda/fato_pedidos_checks.yml
checks for ${SODA_TABLE_NAME}:
  # 1. Validações básicas
  - row_count > 0
  - missing_percent(pedido_id) = 0%
  - missing_percent(data_pedido) = 0%

  # 2. Validações de negócio
  - pedidos_sem_valor:
      pedidos_sem_valor query: |
        SELECT COUNT(*)
        FROM `${SODA_TABLE_NAME}`
        WHERE valor_total IS NULL OR valor_total = 0
      warn: when > 10
      fail: when > 100

  # 3. Consistência referencial
  - clientes_inexistentes:
      clientes_inexistentes query: |
        SELECT COUNT(*)
        FROM `${SODA_TABLE_NAME}` f
        LEFT JOIN `${SODA_PROJECT}.${SODA_DATASET}.dim_clientes` c
          ON f.cliente_id = c.cliente_id
        WHERE c.cliente_id IS NULL
      fail: when > 0

Passo 3: Testar Validações¶

# No ambiente DEV, executar DAG manualmente no Airflow
# Verificar logs para confirmar que validações foram executadas
# Exemplo de log esperado:

✅ Validação dinâmica concluída com sucesso
✅ Validação de colunas: 25 colunas verificadas
✅ Validação genérica: diferença de +2.3%
✅ fato_pedidos_checks.yml aprovado com sucesso!

Passo 4: Monitorar Resultados¶

-- Query para verificar últimas validações
SELECT
  created_at,
  table_id,
  message,
  categoria
FROM `driven-strength-374020.lavoro_logs.soda_logs`
WHERE table_id = 'fato_pedidos'
  AND DATE(created_at) = CURRENT_DATE()
ORDER BY created_at DESC;

📈 Monitoramento e Logs¶

Consultar Logs de Validação¶

Ver falhas do dia:

SELECT
  created_at,
  table_id,
  message,
  categoria
FROM `driven-strength-374020.lavoro_logs.soda_logs`
WHERE DATE(created_at) = CURRENT_DATE()
  AND categoria IN ('Erro', 'Alerta')
ORDER BY created_at DESC;

Ver histórico de uma tabela específica:

SELECT
  DATE(created_at) as data,
  COUNT(*) as total_validacoes,
  COUNTIF(categoria = 'Erro') as erros,
  COUNTIF(categoria = 'Alerta') as alertas
FROM `driven-strength-374020.lavoro_logs.soda_logs`
WHERE table_id = 'fato_faturamento'
  AND DATE(created_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
GROUP BY data
ORDER BY data DESC;

Top 10 tabelas com mais erros:

SELECT
  table_id,
  COUNT(*) as total_erros,
  ARRAY_AGG(message ORDER BY created_at DESC LIMIT 3) as ultimas_mensagens
FROM `driven-strength-374020.lavoro_logs.soda_logs`
WHERE categoria = 'Erro'
  AND DATE(created_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
GROUP BY table_id
ORDER BY total_erros DESC
LIMIT 10;

Métricas e KPIs de Qualidade¶

Taxa de aprovação por projeto:

WITH validacoes AS (
  SELECT
    project_id,
    COUNT(*) as total_validacoes,
    COUNTIF(categoria = 'Aprovado') as aprovadas,
    COUNTIF(categoria = 'Erro') as erros,
    COUNTIF(categoria = 'Alerta') as alertas
  FROM `driven-strength-374020.lavoro_logs.soda_logs`
  WHERE DATE(created_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
  GROUP BY project_id
)
SELECT
  project_id,
  total_validacoes,
  aprovadas,
  erros,
  alertas,
  ROUND(aprovadas * 100.0 / total_validacoes, 2) as taxa_aprovacao_pct
FROM validacoes
ORDER BY taxa_aprovacao_pct ASC;

Tendência de qualidade ao longo do tempo:

SELECT
  DATE(created_at) as data,
  COUNT(*) as total_checks,
  COUNTIF(categoria = 'Aprovado') as aprovados,
  COUNTIF(categoria = 'Erro') as erros,
  ROUND(COUNTIF(categoria = 'Aprovado') * 100.0 / COUNT(*), 2) as taxa_aprovacao
FROM `driven-strength-374020.lavoro_logs.soda_logs`
WHERE DATE(created_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
GROUP BY data
ORDER BY data DESC;

💡 Boas Práticas¶

1. Definição de Thresholds (Limites)¶

Seja conservador inicialmente:

# ✅ BOM: Começar com limites apertados
warn: when > 5%      # Alerta cedo para investigar
fail: when > 50%     # Falha apenas em casos graves

# ❌ RUIM: Limites muito permissivos
warn: when > 80%     # Só alertar quando já está crítico
fail: when > 95%     # Quase nunca vai falhar

Ajustar iterativamente:

Comece com thresholds conservadores (warn: 5%, fail: 50%)
Monitore execuções por 1-2 semanas
Analise logs no BigQuery para identificar padrões
Ajuste thresholds baseado em dados reais observados
Reavalie trimestralmente conforme dados evoluem

2. Uso de Partition Filters¶

Performance é crítica:

# ✅ BOM: Sempre usar filtro de partição quando disponível
checks for tabela_particionada:
  - valores_query:
      valores_query query: |
        SELECT COUNT(*)
        FROM `projeto.dataset.tabela`
        WHERE DATE(data_particao) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
          AND valor < 0

# ❌ RUIM: Full table scan (caro e lento)
checks for tabela_particionada:
  - valores_query:
      valores_query query: |
        SELECT COUNT(*)
        FROM `projeto.dataset.tabela`
        WHERE valor < 0  -- Sem filtro de partição!

3. Naming Conventions¶

Nomes claros e descritivos:

# ✅ BOM: Nome explica o que está sendo validado
- email_invalido = 0:
    name: "🚨 Emails com formato inválido"
    email_invalido query: |
      SELECT COUNT(*) FROM tabela WHERE email NOT LIKE '%@%.%'

# ❌ RUIM: Nome genérico
- check1 = 0:
    check1 query: |
      SELECT COUNT(*) FROM tabela WHERE email NOT LIKE '%@%.%'

4. Checks Incrementais¶

Implemente gradualmente:

# Fase 1: Validações básicas
checks for fato_vendas:
  - row_count > 0
  - missing_percent(venda_id) = 0%

# Fase 2: Adicionar validações de negócio
checks for fato_vendas:
  - row_count > 0
  - missing_percent(venda_id) = 0%
  - duplicate_count(venda_id) = 0
  - missing_percent(data_venda) < 5%

# Fase 3: Validações avançadas
checks for fato_vendas:
  # Checks anteriores +
  - valores_negativos = 0:
      valores_negativos query: |
        SELECT COUNT(*) FROM ${SODA_TABLE_NAME}
        WHERE valor_total < 0

5. Documentação de Checks¶

Sempre documente o propósito:

# ✅ BOM: Comentários explicam a regra de negócio
checks for dim_clientes:
  # Regra de negócio: Todo cliente deve ter pelo menos um contato
  # (email OU telefone). Clientes sem contato não podem receber comunicações.
  - cliente_sem_contato:
      cliente_sem_contato query: |
        SELECT COUNT(*)
        FROM ${SODA_TABLE_NAME}
        WHERE (email IS NULL OR email = '')
          AND (telefone IS NULL OR telefone = '')
      warn: when > 10   # Time comercial deve revisar
      fail: when > 100  # Bloqueio se problema sistêmico

6. Ambientes DEV/QA/PRD¶

Comportamento diferente por ambiente:

# No código do SodaValidator
if any(env in self.dataset_id for env in ('_dev', '_qa')):
    # DEV/QA: Não bloqueia pipeline, apenas loga
    if exit_code == 2:
        self.log.warning("⚠️ Validação falhou, mas continuando (DEV/QA)")
else:
    # PRD: Bloqueia pipeline em caso de falha crítica
    if exit_code == 2:
        raise Exception("❌ Validação falhou! Pipeline interrompido.")

🔍 Troubleshooting¶

Problema 1: YAML Syntax Error¶

Erro:

YAML syntax error | while parsing a block mapping

Causa: Indentação incorreta no arquivo YAML.

Solução:

# ❌ ERRADO: Indentação inconsistente
checks for tabela:
- valor = 0:
    valor query: |
  SELECT COUNT(*) FROM tabela

# ✅ CORRETO: Indentação consistente (2 espaços)
checks for tabela:
  - valor = 0:
      valor query: |
        SELECT COUNT(*) FROM tabela

Problema 2: Métricas Não Computadas¶

Erro:

Metrics 'metrica_custom' were not computed for check

Causa: Nome da métrica não corresponde ao nome do check.

Solução:

# ❌ ERRADO: Nomes diferentes
- minha_metrica = 0:
    outra_metrica query: |
      SELECT COUNT(*) FROM tabela

# ✅ CORRETO: Nomes idênticos
- minha_metrica = 0:
    minha_metrica query: |
      SELECT COUNT(*) FROM tabela

Problema 3: Erro de Particionamento¶

Erro:

Cannot query over table without a filter over column 'data_processamento'

Causa: Tabela requer filtro de partição, mas não foi aplicado.

Solução:

O sistema aplica automaticamente, mas você pode verificar:

# Verificar se variáveis de ambiente foram setadas
print(os.environ.get('SODA_ORIGINAL_PARTITION_FILTER'))
print(os.environ.get('SODA_TEMP_PARTITION_FILTER'))

Problema 4: Queries Lentas¶

Problema: Validação de 50 colunas = 50 queries no BigQuery (lento e caro).

Solução: O sistema já usa query única otimizada:

-- Query única para todas as colunas
SELECT
  CASE WHEN COUNT(*) = 0 THEN 0
  ELSE (COUNT(*) - COUNT(coluna1)) * 100.0 / COUNT(*)
  END AS missing_pct_coluna1,

  CASE WHEN COUNT(*) = 0 THEN 0
  ELSE (COUNT(*) - COUNT(coluna2)) * 100.0 / COUNT(*)
  END AS missing_pct_coluna2,

  -- ... todas as outras colunas
FROM `projeto.dataset.tabela`
WHERE filtro_particao;

📚 Estrutura de Arquivos¶

Localização no Repositório¶

dags/
├── operators/
│   ├── soda_operator.py              # Classe SodaValidator principal
│   └── bq_insert_job.py              # Integração com BQInsertJobOperator
│
├── soda/
│   ├── configuration.yml             # Configuração Soda Core (conexão BigQuery)
│   ├── column_mappings.yml           # Templates de validações dinâmicas
│   ├── generic_missing_data_checks.yml  # Checks aplicados a todas as tabelas
│   ├── {tabela}_checks.yml           # Checks específicos por tabela
│   │   ├── dim_cliente_checks.yml
│   │   ├── fato_faturamento_checks.yml
│   │   └── ...
│   ├── sql/                          # Queries SQL customizadas
│   │   ├── check_diferenca_quantidade.sql
│   │   └── check_linhas_duplicadas.sql
│   └── README.md                     # Documentação técnica detalhada

Arquivo de Configuração Principal¶

# dags/soda/configuration.yml
data_source bigquery:
  type: bigquery
  use_query_cache: false
  query_format: standard
  project_id: driven-strength-374020
  dataset: ${SODA_DATASET}            # Variável de ambiente (dinâmica)
  authentication:
    method: application_default       # Usa credenciais do service account
  sampler:
    type: bigquery
    samples_limit: 50                 # Máximo de amostras em caso de falha

🎓 Próximos Passos¶

Expansão Futura¶

Dashboard de Qualidade: Looker Studio mostrando métricas consolidadas
Alertas Proativos: Integração com Slack/Email para alertas em tempo real
Data Profiling: Estatísticas automáticas sobre distribuição de valores
Quality Score: KPI consolidado de qualidade por projeto/tabela
Automated Remediation: Scripts de correção automática para problemas comuns

Como Contribuir¶

Reportar problemas:

Criar issue no GitHub com tag data-quality
Incluir logs do Airflow e exemplo de dados

Sugerir novos checks:

Documentar regra de negócio
Fornecer query SQL de exemplo
Especificar thresholds recomendados

Melhorar documentação:

Adicionar exemplos práticos
Documentar casos de uso específicos
Incluir troubleshooting de novos erros

🔗 Recursos Adicionais¶

Documentação Oficial¶

Repositórios GitHub¶

CDP: GitHub - CDP Soda Implementation
Lakehouse: GitHub - Lakehouse Data Quality

Documentação Técnica Interna¶

Para detalhes técnicos sobre a implementação do SodaValidator e BQInsertJobOperator, consulte:

dags/soda/README.md (documentação técnica completa)
dags/operators/soda_operator.py (código-fonte comentado)

✨ Qualidade de Dados - Soda Core

Última atualização: Janeiro 2025