Modern Data Stack — Projeto Completo

Pipeline de dados de ponta a ponta simulando um e-commerce brasileiro, usando a stack moderna de engenharia de dados.

Arquitetura

┌──────────────────┐     ┌──────────────────────────────────────────────────────────────────┐
│  Source Postgres │     │                   Data Warehouse (Postgres)                       │
│  (Operacional)   │     │                                                                   │
│                  │     │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐        │
│  clientes        │     │  │ LANDING  │  │   RAW    │  │  SILVER  │  │   GOLD   │        │
│  produtos        │────►│  │(extract) │─►│(dbt view)│─►│(dbt tbl) │─►│(dbt tbl) │       │
│  pedidos         │     │  │          │  │          │  │          │  │          │        │
│  itens_pedido    │     │  └──────────┘  └──────────┘  └──────────┘  └──────────┘        │
│  categorias      │     │   dados brutos  tradução PT   limpeza e    dims, facts           │
└──────────────────┘     │   + metadados   → EN + cast   enriquec.    e marts               │
                         └──────────────────────────────────────────────────────────────────┘
                                                    ▲
                                         ┌──────────┴──────────┐
                                         │    Apache Airflow    │
                                         │  (orquestra diário)  │
                                         └─────────────────────┘

Stack

Componente	Tecnologia	Porta
Banco Operacional	PostgreSQL 16	5432
Data Warehouse	PostgreSQL 16	5432
Extração (EL)	Python / Airbyte-style	—
Transformação	dbt-core 1.11.8 + dbt-postgres	—
Orquestração	Apache Airflow 2.9	8080
Documentação dbt	dbt docs serve	8081

Camadas de dados

LANDING

Dados brutos exatamente como vieram do operacional, com colunas de metadados estilo Airbyte:

• _airbyte_raw_id — UUID único por registro
• _airbyte_extracted_at — timestamp da extração
• _airbyte_loaded_at — timestamp do carregamento

RAW (dbt views)

Views sobre o landing que traduzem nomes de colunas do português para inglês e fazem cast explícito de tipos. Sem lógica de negócio.

SILVER (dbt tables)

Dados limpos e padronizados:

• stg_customers — nome em title case, email em lowercase, telefone normalizado, região derivada do estado
• stg_products — enriquecido com categoria, faixas de preço e status de estoque
• stg_orders — status agrupado, colunas de data derivadas, tempo de processamento
• stg_order_items — total do item calculado, filtros de qualidade

GOLD (dbt tables)

Modelos analíticos prontos para consumo por BI / APIs:

Modelo	Descrição
dim_customers	Dimensão cliente com segmentação RFM e histórico de compras
dim_products	Dimensão produto com métricas de vendas e popularidade
fct_orders	Fato pedidos (tabela central do modelo dimensional)
mart_daily_sales	Vendas diárias com acumulados mensais e taxas
mart_customer_ltv	LTV, scores RFM (1-5) e segmentação por cliente
mart_product_performance	Receita, ranking e curva ABC por produto

Pré-requisitos

• Docker Desktop (ou Docker Engine + Compose)
• 4 GB de RAM disponível para os containers

Como rodar

1. Clone o repositório

git clone https://github.com/<seu-usuario>/modern-data-stack.git
cd modern-data-stack

2. Configure o ambiente

cp .env.example .env

O .env.example já vem com os valores padrão para desenvolvimento local. Você só precisa gerar a AIRFLOW__CORE__FERNET_KEY:

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

3. Suba todos os serviços

make up
# ou:
docker compose up -d --build

Aguarde ~60 segundos para todos os serviços inicializarem.

4. Instale os pacotes dbt

make dbt-install

5. Execute o pipeline manualmente (primeira vez)

# Extração: source → landing
make airbyte-extract

# Transformações dbt (camada a camada)
make dbt-run-raw
make dbt-run-silver
make dbt-run-gold

# Testes de qualidade
make dbt-test

Ou rode tudo de uma vez pelo Airflow:

• Acesse http://localhost:8080 (admin / admin)
• Ative a DAG modern_data_stack_pipeline
• Clique em Trigger DAG

6. Acesse os dados

# Banco operacional
make psql-source

# Data Warehouse
make psql-warehouse

Exemplo de query no warehouse:

-- Top 10 clientes por LTV
SELECT customer_name, rfm_segment, net_monetary, estimated_annual_ltv
FROM gold.mart_customer_ltv
ORDER BY net_monetary DESC
LIMIT 10;

-- Receita diária dos últimos 30 dias
SELECT order_date, gross_revenue, net_revenue, total_orders, cancellation_rate_pct
FROM gold.mart_daily_sales
WHERE order_date >= current_date - 30
ORDER BY order_date;

-- Curva ABC de produtos
SELECT abc_curve, count(*) as produtos, sum(total_net_revenue) as receita
FROM gold.mart_product_performance
GROUP BY 1 ORDER BY 1;

7. Documentação dbt

make dbt-docs
# Acesse: http://localhost:8081

Agendamento (Airflow)

A DAG modern_data_stack_pipeline roda automaticamente todo dia às 02:00 BRT.

Grafo de dependências:

check_connectivity
       │
 extract_to_landing
       │
  dbt_run_raw ──► dbt_test_raw
       │
 dbt_run_silver ──► dbt_test_silver
       │
  dbt_run_gold ──► dbt_test_gold
       │
 dbt_docs_generate

Estrutura do projeto

modern-data-stack/
├── docker-compose.yml
├── .env.example             # Copie para .env antes de rodar
├── Makefile
├── postgres/
│   ├── init.sh
│   ├── source/init.sql          # DDL + dados operacionais
│   └── warehouse/init.sql       # Schemas landing/raw/silver/gold
├── scripts/                     # Setup alternativo sem Docker
│   ├── 01_create_databases.sql
│   ├── 02_populate_source_db.sql
│   └── 03_setup_warehouse.sql
├── airbyte/
│   └── extract.py               # Extração source → landing (Airbyte-style)
├── dbt/
│   ├── dbt_project.yml
│   ├── profiles.yml
│   ├── packages.yml
│   ├── macros/
│   │   └── generate_schema_name.sql
│   ├── models/
│   │   ├── sources.yml          # Fonte: schema landing
│   │   ├── raw/                 # Views: tradução PT→EN + cast de tipos
│   │   ├── silver/              # Tabelas: limpeza e padronização
│   │   └── gold/                # Tabelas: dims, facts e marts analíticos
│   └── tests/singular/          # Testes de negócio customizados
└── airflow/
    ├── Dockerfile
    ├── requirements.txt
    └── dags/
        └── modern_data_stack_dag.py

Substituindo pelo Airbyte real

O script airbyte/extract.py replica o que o Airbyte faz internamente. Para usar o Airbyte real em produção:

1. Instale o Airbyte (via Docker ou Airbyte Cloud)
2. Configure uma conexão source-postgres apontando para source_postgres
3. Configure um destination-postgres apontando para warehouse_postgres, schema landing
4. Substitua a task extract_to_landing na DAG pelo AirbyteTriggerSyncOperator

from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator

extract_to_landing = AirbyteTriggerSyncOperator(
    task_id="extract_to_landing",
    airbyte_conn_id="airbyte_default",
    connection_id="<seu-connection-id>",
    asynchronous=False,
)