Convenções de Dados - Data Lake Lavoro¶

Este guia documenta todos os padrões e convenções para nomenclatura, estrutura de tabelas, tipos de dados e modelagem utilizados no Data Lake Lavoro.

📐 Nomenclatura¶

Datasets (BigQuery)¶

Padrão: {camada}_{projeto}_{ambiente?}

Os datasets seguem o padrão de 1 dataset por camada, projeto e ambiente. Como temos 1 único Composer (Lakehouse) orquestrando todos os 4 projetos, os "projetos" existem apenas a nível de dataset, e cada um possui seus datasets isolados por ambiente.

Ambientes:

DEV: Desenvolvimento - sufixo _dev
QA: Quality Assurance - sufixo _qa
PRD: Produção - sem sufixo

Exemplos:

# Ambiente DEV
bronze_operacoes_dev
silver_operacoes_dev
gold_operacoes_dev

# Ambiente QA
bronze_lakehouse_qa
silver_lakehouse_qa
gold_lakehouse_qa

# Ambiente PRD (sem sufixo)
bronze_pessoas
silver_pessoas
gold_pessoas

Regras¶

Regra	Descrição
✅ Sempre em lowercase	Todos os caracteres em minúsculas
✅ Separador `_` (underscore)	Separar camada, projeto e ambiente com underscore
✅ Sem espaços ou caracteres especiais	Apenas letras, números e underscore
✅ Formato: `camada_projeto_ambiente`	Exemplo: `bronze_operacoes_dev`, `bronze_operacoes` (PRD)
✅ PRD sem sufixo	Produção não possui sufixo de ambiente

Tabelas¶

🥉 Bronze (Zona de Entrada)¶

A camada Bronze contém dados brutos extraídos dos sistemas fonte. O padrão de nomenclatura varia conforme o tipo de extração:

Padrão para extração de tabelas: {empresa}_{sistema}_{tabela}

Exemplos:

lavoro_sap_zc_gllitems
lavoro_sap_oitm
desempar_protheus_sd2010
desempar_protheus_sc7010
nova_geracao_sapb1_oitm

Padrão para extração baseada em query/objeto: {empresa}_{sistema}_{objeto}

Exemplos:

lavoro_sap_faturamento
desempar_protheus_pedidos_de_compra
nova_geracao_sapb1_balancete

🥈 Silver (Zona Validada)¶

Padrão: Ainda mantém o prefixo da empresa para rastreabilidade

Exemplos:

lavoro_sap_faturamento
desempar_protheus_pedidos
nova_geracao_sapb1_clientes

🥇 Gold (Zona de Consumo)¶

Dimensões: dim_{entidade} Fatos: fato_{entidade}

Exemplos de Dimensões:

dim_estrutura_comercial
dim_clientes
dim_produtos
dim_fornecedores

Exemplos de Fatos:

fato_faturamento
fato_estoque
fato_pedidos_de_compra
fato_balancete

Regras¶

Regra	Descrição
✅ Sempre em lowercase	Todos os caracteres em minúsculas
✅ Separador `_` (underscore)	Separar palavras com underscore
✅ Nomes descritivos	Evitar abreviações, ser claro e objetivo
✅ Bronze: `empresa_sistema_tabela/objeto`	Identifica origem dos dados
✅ Gold: `dim_` ou `fato_`	Prefixo indica tipo de tabela dimensional

DAGs, Task Groups, Tasks e Templates SQL¶

📋 DAGs¶

DAGs (Directed Acyclic Graphs) são os workflows principais que organizam as execuções de pipelines.

Padrão Bronze e Silver:

Formato: {empresa}_{sistema}
Regra: Bronze e Silver sempre na mesma DAG
Exemplos:
lavoro_sap
desempar_protheus
nova_geracao_sapb1

Padrão Gold:

Formato: {fato/dim}_{objeto}
Regra: Organizadas por fato ou dimensão
Consolidação: Em alguns casos, agrupamos tabelas do mesmo tema
Exemplo: DAG fato_faturamento agrupa fato_faturamento e visao_faturamento
Exemplos:
fato_faturamento
dim_estrutura_comercial

📦 Task Groups¶

Task Groups agrupam tarefas relacionadas dentro de uma DAG.

Padrão Bronze e Silver:

Formato: {empresa}_{sistema}_{camada}
Exemplos:
desempar_protheus_bronze
desempar_protheus_silver

Padrão Gold:

Formato: {fato/dim}_{objeto}_{camada}
Exemplos:
fato_faturamento_gold
dim_fornecedores_gold

⚙️ Tasks¶

Tasks são as unidades individuais de trabalho dentro dos Task Groups. Seguem o mesmo modelo, substituindo a camada pelo nome da tabela/objeto.

Padrão Bronze e Silver:

Formato: {empresa}_{sistema}_{tabela/objeto}
Exemplos:
desempar_protheus_pedidos_de_compra
lavoro_sap_faturamento

Padrão Gold:

Formato: {fato/dim}_{objeto}_{tabela}
Exemplos:
fato_faturamento_historico
dim_produtos_atualizar

📄 Templates SQL¶

Templates SQL são arquivos .sql reutilizáveis que contêm as queries executadas pelas tasks.

Padrão:

Formato: {empresa}_{sistema}_{objeto}.sql
Localização: /dags/task_groups/{camada}/sql/

Otimização com Jinja:

Quando múltiplas empresas usam o mesmo sistema, criamos um único template e manipulamos via Jinja para reutilização.

Exemplo: protheus_faturamento.sql (usado por Desempar, Futuragro, etc.)
Benefício: Variáveis Jinja substituem empresa/configurações específicas
Vantagem: Reduz duplicação de código e facilita manutenção

Observação: O projeto Lakehouse possui padrões próprios de nomenclatura.

🎯 Exemplo End-to-End de Nomenclatura¶

Para ilustrar como todas as convenções se conectam na prática, vamos seguir um exemplo completo desde a extração Bronze até a camada Gold:

Cenário: Extração de Pedidos de Compra da Desempar (sistema Protheus)¶

1. DAG (Bronze + Silver juntos)

Arquivo: dags/desempar_protheus.py
DAG ID: desempar_protheus

2. Task Group Bronze

# Arquivo: dags/task_groups/bronze/desempar_protheus_bronze.py
# Task Group ID: desempar_protheus_bronze

with TaskGroup('desempar_protheus_bronze', tooltip='Extração Bronze Protheus') as bronze_group:
    # Task para extrair pedidos de compra
    # Task ID: desempar_protheus_pedidos_de_compra
    extract_pedidos = PythonOperator(
        task_id='desempar_protheus_pedidos_de_compra',
        python_callable=extract_from_protheus,
        op_kwargs={'table': 'SC7010'}  # Tabela source no Protheus
    )

Tabela Bronze criada:

Dataset: bronze_operacoes
Tabela: desempar_protheus_pedidos_de_compra
Padrão: {empresa}_{sistema}_{tabela}

3. Task Group Silver

# Arquivo: dags/task_groups/silver/desempar_protheus_silver.py
# Task Group ID: desempar_protheus_silver

with TaskGroup('desempar_protheus_silver', tooltip='Transformação Silver Protheus') as silver_group:
    # Task para transformar pedidos
    # Task ID: desempar_protheus_pedidos
    transform_pedidos = BQInsertJobOperator(
        task_id='desempar_protheus_pedidos',
        configuration={
            'query': {
                'query': open('task_groups/silver/sql/protheus_pedidos.sql').read(),
                'useLegacySql': False
            }
        }
    )

Template SQL Silver (reutilizável via Jinja):

-- Arquivo: dags/task_groups/silver/sql/protheus_pedidos.sql
-- Template reutilizável para qualquer empresa Protheus

SELECT
    CONCAT(LPAD(C7_NUM, 18, '#'), '-PROTHEUS') as pedido_id,
    UPPER(TRIM(C7_FORNECE)) as fornecedor_codigo,
    UPPER(TRIM(C7_PRODUTO)) as produto_codigo,
    SAFE_CAST(C7_QUANT AS FLOAT64) as quantidade,
    SAFE_CAST(C7_PRECO AS NUMERIC(38,2)) as valor_unitario,
    PARSE_DATE('%Y%m%d', C7_EMISSAO) as data_emissao,
    UPPER(TRIM(C7_USER)) as usuario,
    extraction_timestamp
FROM {{ params.bronze_dataset }}.{{ params.bronze_table }}
WHERE D_E_L_E_T_ = ''  -- Protheus soft delete flag

Tabela Silver criada:

Dataset: silver_operacoes
Tabela: desempar_protheus_pedidos
Padrão: {empresa}_{sistema}_{objeto}

4. DAG Gold

Arquivo: dags/fato_pedidos_de_compra.py
DAG ID: fato_pedidos_de_compra

5. Task Group Gold

# Arquivo: dags/task_groups/gold/fato_pedidos_de_compra_gold.py
# Task Group ID: fato_pedidos_de_compra_gold

with TaskGroup('fato_pedidos_de_compra_gold', tooltip='Fato Pedidos Gold') as gold_group:
    # Task para criar fato consolidado
    # Task ID: fato_pedidos_de_compra_consolidado
    create_fato = BQInsertJobOperator(
        task_id='fato_pedidos_de_compra_consolidado',
        configuration={
            'query': {
                'query': open('task_groups/gold/sql/fato_pedidos_de_compra.sql').read(),
                'useLegacySql': False
            }
        }
    )

Template SQL Gold:

-- Arquivo: dags/task_groups/gold/sql/fato_pedidos_de_compra.sql
-- Consolida pedidos de múltiplas empresas

SELECT
    pedido_id,
    fornecedor_id,  -- FK para dim_fornecedores
    produto_id,     -- FK para dim_produtos
    data_emissao,   -- FK para dim_calendario
    quantidade,
    valor_unitario,
    quantidade * valor_unitario as valor_total,
    'DESEMPAR' as empresa
FROM silver_operacoes.desempar_protheus_pedidos

UNION ALL

SELECT
    pedido_id,
    fornecedor_id,
    produto_id,
    data_emissao,
    quantidade,
    valor_unitario,
    quantidade * valor_unitario as valor_total,
    'FUTURAGRO' as empresa
FROM silver_operacoes.futuragro_protheus_pedidos

Tabela Gold criada:

Dataset: gold_operacoes
Tabela: fato_pedidos_de_compra
Padrão: fato_{entidade}

Fluxo Completo Resumido¶

graph LR
    A[DAG: desempar_protheus] --> B[Task Group: desempar_protheus_bronze]
    B --> C[Task: desempar_protheus_pedidos_de_compra]
    C --> D[Tabela: bronze_operacoes.desempar_protheus_pedidos_de_compra]

    D --> E[Task Group: desempar_protheus_silver]
    E --> F[Task: desempar_protheus_pedidos]
    F --> G[SQL: protheus_pedidos.sql]
    G --> H[Tabela: silver_operacoes.desempar_protheus_pedidos]

    H --> J[Task Group: fato_pedidos_de_compra_gold]
    I[DAG: fato_pedidos_de_compra] --> J
    J --> K[Task: fato_pedidos_de_compra_consolidado]
    K --> L[SQL: fato_pedidos_de_compra.sql]
    L --> M[Tabela: gold_operacoes.fato_pedidos_de_compra]

    style D fill:#cd7f32,color:#fff
    style H fill:#c0c0c0
    style M fill:#ffd700

Benefícios desta Estrutura¶

Rastreabilidade: Nome completo indica origem (empresa → sistema → objeto)
Reutilização: Template SQL protheus_pedidos.sql serve para Desempar, Futuragro, etc.
Consistência: Todos os nomes seguem mesmo padrão
Manutenção: Fácil localizar arquivos e entender fluxo
Documentação: Estrutura auto-explicativa

Colunas¶

Padrão: {descricao}_{tipo?}

Importante: Todas as extrações geram automaticamente uma coluna extraction_timestamp com o timestamp da extração.

Regras¶

Regra	Descrição
✅ Sempre em lowercase (snake_case)	Todos os caracteres em minúsculas
✅ Nomes descritivos e autoexplicativos	Nome deve deixar claro o conteúdo
✅ Evitar abreviações genéricas	Preferir nomes completos: `quantidade` ao invés de `qtd`
✅ Exceções permitidas	Apenas sufixo `_id` e siglas conhecidas (`cpf`, `cnpj`, `cnae`)
✅ Sufixo para indicar tipo	Quando necessário para clareza
✅ Coluna obrigatória: `extraction_timestamp`	Timestamp automático da extração

Sufixos Padronizados¶

A única abreviação aceita é _id para identificadores únicos. Todos os outros nomes devem ser escritos por extenso:

Sufixo	Tipo	Exemplo	Justificativa
`_id`	Identificador único	`pedido_id`, `cliente_id`, `produto_id`	Padrão universal em bancos de dados
`_key`	Chave de negócio	`pedido_key`, `cliente_key`	Convenção de data warehousing
`_at` / `_em`	Timestamp	`created_at`, `updated_at`, `deleted_at`	Padrão de frameworks (Ruby on Rails)
`_date` / `_data`	Date	`data_pedido`, `data_nascimento`	Indicador claro de tipo DATE
`_flg` / `_flag`	Boolean	`ativo_flg`, `is_deleted`	Convenção de modelagem dimensional
`_valor`	Valor monetário	`valor_total`, `valor_liquido`	SEMPRE POR EXTENSO - não use `amt`
`_quantidade`	Quantidade	`quantidade_itens`, `quantidade_vendida`	SEMPRE POR EXTENSO - não use `qty` ou `qtd`
`_percentual`	Percentual	`percentual_desconto`, `taxa_conversao_percentual`	SEMPRE POR EXTENSO - não use `pct` ou `perc`

Importante: Com exceção de _id, todas as abreviações devem ser evitadas. Escreva sempre o nome completo em português para garantir clareza e consistência.

Exemplos de Colunas Bem Nomeadas¶

-- ✅ BOM
pedido_id               -- Identificador único do pedido
cliente_id              -- FK para dimensão de clientes
data_pedido             -- Data do pedido
valor_total             -- Valor total do pedido
valor_desconto          -- Valor de desconto aplicado
quantidade_items        -- Quantidade de items
status                  -- Status atual do pedido
created_at              -- Timestamp de criação
updated_at              -- Timestamp de última modificação
is_valid                -- Flag de validação
ativo_flg               -- Flag se registro está ativo

-- ❌ RUIM
ped_id                  -- Abreviação desnecessária
dt_ped                  -- Abreviação + não descritivo
vlr_tot                 -- Difícil de entender
qtd                     -- Sem contexto
flg_ativo               -- Prefixo desnecessário

🎯 Tipos de Dados (Data Types)¶

Padrão de IDs¶

Todos os IDs devem ser formatados utilizando LPAD para garantir consistência e padronização, utilizando o sistema no final.

Regra: Completar 18 caracteres no total usando # como caractere de preenchimento.

Formato: CONCAT(LPAD(id, 18, '#') - SISTEMA)

Exemplos:

-- ID com código da empresa/sistema como sufixo
CONCAT(LPAD('789', 18, '#'),'-', SISTEMA)         -- Resultado: ###########789-SAP

-- ID aplicado em query
SELECT
  CONCAT(LPAD('789', 18, '#'),'-', SISTEMA) ) as produto_id
FROM bronze_operacoes.lavoro_sap_material

Benefícios:

✅ IDs uniformes e previsíveis
✅ Facilita ordenação alfabética
✅ Identifica origem do dado pelo sufixo
✅ Evita colisões entre sistemas diferentes
✅ Comprimento fixo de 18 caracteres

Mapeamento de Tipos BigQuery¶

Tipo de Dado	BigQuery Type	Exemplo	Notas
Identificadores	`STRING`	`cliente_id: 'C12345'`	UUIDs, códigos, chaves
Textos curtos	`STRING`	`nome: 'João Silva'`	Nomes, descrições curtas
Textos longos	`STRING`	`descricao: '...'`	Descrições, comentários
Números inteiros	`INT64`	`quantidade: 10`	Contadores, quantidades
Decimais	`FLOAT64`	`valor: 99.90`	Valores monetários, percentuais
Monetário (precisão)	`NUMERIC(38, 2)`	`valor_total: 1234.56`	Valores financeiros precisos
Datas	`DATE`	`data_pedido: '2025-01-15'`	Apenas data (sem hora)
Timestamps	`TIMESTAMP`	`created_at: '2025-01-15 14:30:00 UTC'`	Data e hora com timezone
Hora do dia	`TIME`	`hora_abertura: '09:00:00'`	Apenas hora (sem data)
Booleanos	`BOOLEAN`	`is_active: true`	Flags, indicadores
JSON	`JSON`	`metadata: {"key": "value"}`	Dados semi-estruturados
Arrays	`ARRAY<T>`	`tags: ['venda', 'online']`	Listas de valores
Structs	`STRUCT<...>`	`endereco: {rua, cidade, estado}`	Objetos aninhados
Geografia	`GEOGRAPHY`	`localizacao: POINT(-23.55, -46.63)`	Dados geoespaciais

Regras de Casting e Conversão¶

Datas e Timestamps¶

-- ✅ BOM: Parse explícito com formato
PARSE_DATE('%Y-%m-%d', '2025-01-15')
PARSE_TIMESTAMP('%Y-%m-%d %H:%M:%S', '2025-01-15 14:30:00')

-- ✅ BOM: Conversão segura com validação
SAFE.PARSE_DATE('%Y-%m-%d', data_string)

-- ✅ BOM: Formato ISO 8601
CAST('2025-01-15' AS DATE)
CAST('2025-01-15T14:30:00Z' AS TIMESTAMP)

-- ❌ RUIM: Conversão implícita sem validação
data_string  -- Pode falhar em runtime

Números¶

-- ✅ BOM: Cast explícito com validação
SAFE_CAST(valor_string AS FLOAT64)
CAST(quantidade_string AS INT64)

-- ✅ BOM: Tratamento de nulos
COALESCE(SAFE_CAST(valor AS FLOAT64), 0.0)

-- ❌ RUIM: Assumir tipo sem validar
valor_string * 1.0  -- Pode falhar

Strings¶

Importante: Todas as strings devem ser armazenadas em UPPER CASE (maiúsculas).

-- ✅ BOM: Normalização de strings (sempre UPPER)
UPPER(TRIM(nome))  -- Maiúsculas e sem espaços

-- ✅ BOM: Emails em lowercase (exceção)
LOWER(TRIM(email))  -- Emails sempre lowercase

-- ✅ BOM: Remoção de caracteres especiais
REGEXP_REPLACE(cpf, r'[^0-9]', '')  -- Apenas números

-- ✅ BOM: Validação de formato
REGEXP_CONTAINS(email, r'^[^@]+@[^@]+\.[^@]+$')  -- Valida email

-- ❌ RUIM: Strings sem normalização
nome  -- Pode ter espaços extras e case inconsistente

📊 Particionamento¶

O que é uma Tabela Particionada?¶

Uma tabela particionada é dividida em segmentos menores (partições) baseados em uma coluna específica, geralmente uma data. Isso melhora significativamente a performance e reduz custos ao processar apenas as partições necessárias.

Quando Particionar¶

Particione tabelas que atendam qualquer critério:

Critério	Descrição
✅ Tabela > 1 GB	Tabelas grandes se beneficiam de particionamento
✅ Queries filtram por data	Quando a maioria das queries filtra por data
✅ Granularidade temporal	Dados diários, mensais ou com timestamp
✅ Dados históricos	Quando salvamos dados históricos (ex: `fato_faturamento_historico`)

Exemplo de uso: Na camada Gold, quando criamos tabelas históricas como fato_faturamento_historico, particionamos por data para facilitar consultas de períodos específicos e melhorar a performance.

Regras de Particionamento¶

Regra	Descrição
✅ Sempre por dia	Particionamento diário é o padrão
✅ Use coluna de data quando disponível	Preferir `PARTITION BY data_coluna`
✅ Converta timestamp para data	Use `DATE(timestamp_coluna)` quando necessário
✅ Documente a coluna de partição	Deixe claro qual coluna particiona a tabela

🏗️ Modelagem de Dados¶

Star Schema (Esquema Estrela)¶

O Star Schema é o modelo dimensional mais utilizado em data warehouses. Ele organiza os dados em duas categorias principais: Fatos e Dimensões.

O que é Star Schema?¶

O Star Schema recebe esse nome porque visualmente se assemelha a uma estrela: uma tabela de fatos central conectada a várias tabelas de dimensões ao redor. Esse modelo facilita:

✅ Consultas analíticas rápidas
✅ Agregações e cálculos de métricas
✅ Relacionamentos claros entre entidades
✅ Performance otimizada para BI

Tabelas de Fatos¶

Fatos armazenam as métricas e medidas do negócio - valores numéricos que podem ser agregados e analisados.

Características: - Contêm métricas quantitativas (valores, quantidades, percentuais) - Possuem chaves estrangeiras (FKs) para tabelas de dimensões - Geralmente são grandes em volume de registros - Registram eventos ou transações do negócio

Tipos de métricas: - Aditivas: Podem ser somadas em qualquer dimensão (ex: valor_total, quantidade) - Semi-aditivas: Podem ser somadas em algumas dimensões, mas não em todas (ex: saldo) - Não-aditivas: Não podem ser somadas (ex: percentuais, taxas)

Exemplos: fato_faturamento, fato_estoque, fato_pedidos_de_compra

Tabelas de Dimensões¶

Dimensões fornecem o contexto para os fatos - descrevem o "quem", "o quê", "onde", "quando" e "como" das métricas.

Características: - Contêm atributos descritivos e textuais - Geralmente menores em volume que tabelas de fatos - Podem conter hierarquias (ex: produto → categoria → departamento) - Possibilitam filtros e agrupamentos nas análises

Versionamento (SCD - Slowly Changing Dimensions): - Type 1: Sobrescreve valores antigos (sem histórico) - Type 2: Mantém histórico completo com versões - Type 3: Mantém valor anterior e atual

Exemplos: dim_clientes, dim_produtos, dim_estrutura_comercial, dim_fornecedores

Como Trabalhar com Fatos e Dimensões¶

Relacionamentos¶

Fatos se relacionam com dimensões através de chaves estrangeiras (FKs)
Uma tabela de fatos pode ter múltiplas dimensões
Dimensões podem ser compartilhadas entre diferentes fatos (dimensões conformadas)

Desnormalização¶

Dimensões são geralmente desnormalizadas para melhorar performance
Atributos relacionados ficam na mesma tabela (ex: endereço completo dentro de dim_clientes)
Reduz número de JOINs necessários nas consultas

Granularidade¶

Definir a granularidade (nível de detalhe) dos fatos é crítico
Exemplo: fato_faturamento pode ter granularidade de item (1 linha por item de nota) ou nota (1 linha por nota fiscal)
Granularidade mais baixa = maior flexibilidade, mas maior volume de dados

✅ Data Quality Checks¶

A Lavoro utiliza Soda Core integrado aos pipelines Airflow para garantir qualidade de dados em todas as camadas (Bronze, Silver e Gold).

Abordagens de Validação¶

Soda Core (validações em tempo de execução)
Validações dinâmicas (email, freshness, CPF/CNPJ)
Validações de missing data (auto-geradas para todas as colunas)
Validações genéricas (duplicatas, contagem de registros)
Validações customizadas por tabela (regras de negócio específicas)
DAG de Validação de Cargas (freshness monitoring)
Monitoramento de atualidade dos dados
Alertas quando tabelas não são atualizadas no SLA
123 tabelas monitoradas (93 Silver + 30 Gold)

Importante: Os checks podem ou não impedir a atualização completa das tabelas/DAGs, dependendo da criticidade e ambiente (DEV/QA vs PRD).

📖 Documentação Completa¶

Para detalhes sobre implementação, arquitetura, tipos de validação, monitoramento e como adicionar novas validações, consulte:

→ Qualidade de Dados - Documentação Completa

A documentação completa inclui: - 🏗️ Arquitetura e integração com Airflow - ✅ 4 tipos de validação detalhados com exemplos - 📊 Tabelas monitoradas por projeto - 🔧 Guia passo a passo para adicionar validações - 📈 Queries SQL para monitoramento - 💡 Boas práticas e troubleshooting

Dimensões de Qualidade Monitoradas¶

Dimensão	Ferramenta	Exemplo de Validação
Completude	Soda Core	Missing data < 50%, row count > 0
Unicidade	Soda Core	Duplicate count = 0, unique IDs
Validade	Soda Core	Email format válido, outliers detectados, ranges de valores
Atualidade (Freshness)	DAG validacao_cargas + Soda Core	Última atualização < 24h, freshness por coluna

Monitoramento de Freshness (validacao_cargas.py)¶

Objetivo: Monitorar se as tabelas estão sendo atualizadas conforme esperado, detectando dados desatualizados antes que usuários de negócio identifiquem o problema.

Como Funciona¶

O DAG validacao_cargas.py executa diariamente e verifica se as tabelas críticas de Silver e Gold foram atualizadas nas últimas 24 horas.

Lógica de validação:

-- Para cada tabela monitorada, verifica timestamp da última atualização
SELECT
    table_name,
    MAX(extraction_timestamp) as ultima_atualizacao,
    CURRENT_TIMESTAMP() as agora,
    TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), MAX(extraction_timestamp), HOUR) as horas_desde_atualizacao
FROM `{dataset}.{table}`
WHERE DATE(extraction_timestamp) >= CURRENT_DATE() - 2
GROUP BY table_name
HAVING horas_desde_atualizacao > 24  -- Alerta se > 24h sem atualização

Critérios de Alerta¶

Critério	Ação	Destinatário
Dados atualizados D-1	✅ Nenhuma ação	-
Dados desatualizados > 24h	⚠️ Envia alerta por e-mail	Time de dados
Dados desatualizados > 48h	❌ Alerta crítico + notificação Slack	Time de dados + Gestores

Tabelas Monitoradas¶

Silver (93 tabelas): - Todas as tabelas de clientes: *_cliente - Todas as tabelas de vendas: *_faturamento, *_pedidos - Tabelas críticas de estoque: *_estoque - Dados contábeis: *_balancete, *_contas_contabeis

Gold (30 tabelas): - Todas as dimensões: dim_* - Fatos críticos: fato_faturamento, fato_estoque, fato_pedidos_de_compra

Como Adicionar Nova Tabela ao Monitoramento¶

Editar o DAG:

# dags/validacao_cargas.py

# Lista de tabelas a monitorar
TABELAS_MONITORADAS = [
    {'dataset': 'silver_cdp', 'table': 'lavoro_sap_clientes'},
    {'dataset': 'silver_cdp', 'table': 'desempar_protheus_clientes'},
    {'dataset': 'gold_cdp', 'table': 'dim_clientes'},
    {'dataset': 'gold_operacoes', 'table': 'fato_faturamento'},
    # Adicionar nova tabela aqui
    {'dataset': 'silver_operacoes', 'table': 'nova_tabela'},
]

Definir SLA customizado (opcional):

# Para tabelas com SLA diferente de 24h
CUSTOM_SLA = {
    'gold_lakehouse.fato_faturamento': 8,  # 8h (atualização 3x ao dia)
    'silver_lakehouse.lavoro_sap_faturamento': 4,  # 4h (atualização frequente)
}

Exemplo de E-mail de Alerta¶

Assunto: [ALERTA] Tabelas desatualizadas no Data Lake

Olá time de dados,

As seguintes tabelas não foram atualizadas nas últimas 24 horas:

🔴 CRÍTICO (> 48h):
- silver_operacoes.qualicitrus_hpro_estoque
  Última atualização: 2025-01-20 08:00:00 (52 horas atrás)

⚠️ WARNING (> 24h):
- gold_cdp.dim_clientes
  Última atualização: 2025-01-21 15:00:00 (29 horas atrás)
- silver_cdp.desempar_protheus_pedidos
  Última atualização: 2025-01-21 14:30:00 (29.5 horas atrás)

Ações recomendadas:
1. Verificar execução dos DAGs no Airflow
2. Verificar logs de erro
3. Validar conectividade com sistemas fonte

Dashboard de monitoramento: [link]
Airflow: [link]

Agendamento¶

Schedule: Diariamente às 08:00 (após execução dos DAGs principais)

schedule_interval='0 8 * * *'

Rational: Executar após a janela de atualização dos DAGs (06:00) para validar se dados foram carregados corretamente.

Benefícios¶

✅ Detecção proativa: Identifica problemas antes que usuários reportem
✅ Redução de incidentes: Alerta médio 6 horas antes de impactar dashboards
✅ Cobertura ampla: 123 tabelas monitoradas (93 Silver + 30 Gold)
✅ SLA garantido: Freshness < 2 dias assegurado
✅ Rastreabilidade: Histórico de alertas para análise de tendências

📝 Documentação de Tabelas¶

Obrigatoriedade de Documentação¶

Toda tabela nas camadas Silver e Gold deve obrigatoriamente ter:

✅ Descrição da tabela
✅ Descrição de todas as colunas

Método: BQInsertJobOperator¶

Utilizamos o operator BQInsertJobOperator do Airflow para documentar tabelas através de dicionários que geram os comandos SQL automaticamente.

Parâmetros utilizados:

table_description: Descrição da tabela
schema_fields_updates: Lista de dicionários com descrições das colunas

Exemplo de Implementação¶

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

# Dicionário com descrições
table_metadata = {
    'table_description':
        'Tipo de extração: "full".\n'
        'Tecnologia da fonte de dados: SQL Server.\n'
        'Banco/Tabelas: DBPRODDES.DBO.CT2010, DBPRODDES.DBO.CT1010, DBPRODDES.DBO.SX5010',

    'schema_fields_updates': [
        {
            "name": "conta_id",
            "description": "Campo origem: COD_CONTA, que é CT2_DEBITO ou CT2_CREDIT."
        },
        {
            "name": "conta_descricao",
            "description": "Campo origem: DESC_CONTA, que é CT1_DESC01. Espaços em branco removidos."
        },
        {
            "name": "valor",
            "description": "Campo origem: VALOR, que é SUM(CT2_VALOR) ou CT2_VALOR. Valor arredondado para duas casas decimais ou valor individual."
        },
        {
            "name": "natureza_contabil",
            "description": "Campo baseado no tipo de lançamento: 'DEBITO' ou 'CREDITO'."
        },
        {
            "name": "data_criacao",
            "description": "Campo origem: DATA, que é '20281231' ou CT2_DATA."
        }
    ]
}

Modelo de Descrições¶

Descrição da tabela:

Deve explicar o tipo de extração (full, incremental, query)
Indicar a tecnologia da fonte de dados (SQL Server, SAP, etc.)
Listar banco/tabelas de origem

Descrição das colunas:

Indicar campo de origem quando aplicável
Explicar transformações realizadas
Ser clara e objetiva sobre o conteúdo do campo