🏗️ Pipeline Ecommerce — Documentação Técnica

🚀 0. Jornada de um Dado — Do Pedido ao KPI

O cenário: cliente faz um pedido

Um cliente chamado João Silva (customer_id=42) faz um pedido de R$299,90 às 14:00 de uma terça-feira. Veja o que acontece com esse dado em cada camada:

📦 Origem — PostgreSQL OLTP (tempo real)

O sistema de e-commerce grava imediatamente na tabela ecommerce.orders:

🗃️ Camada Raw — MinIO (extração a cada hora)

O DAG ecommerce_oltp_to_raw roda às 15:00 e detecta o pedido via watermark (updated_at > '14:00'). Salva o dado exatamente como veio do banco, sem nenhuma alteração:

s3://raw/ecommerce/orders/
  year=2026/month=03/day=16/hour=15/
    run_2026-03-16T15-00-00.parquet
# → 1 linha, idêntica ao OLTP

🥉 Camada Bronze — Validação Pydantic

O DAG ecommerce_raw_to_bronze lê o Parquet e passa cada linha pelo schema OrderSchema. O pedido é válido → vai para o bucket bronze com 2 colunas extras de rastreabilidade:

🥈 Camada Silver — Delta Lake (UPSERT)

O DAG ecommerce_bronze_to_silver faz MERGE no Delta Lake. Como o order_id=1001 não existe ainda, é um INSERT:

🔄 Silver — Depois: o pedido é enviado (UPDATE)

3 horas depois, o pedido muda de pending → shipped no OLTP. Na próxima extração às 18:00, a pipeline detecta a mudança (watermark) e faz MERGE:

🧬 Silver — SCD Type 2 para clientes (o mais importante!)

João Silva atualiza o email dele de joao@gmail.com → joao@empresa.com. Na Silver, o histórico é preservado:

Resultado: você pode responder "qual era o email do João quando ele fez o pedido em 2025?" — basta fazer JOIN pelo _valid_from / _valid_to.

🥇 Camada Gold — Star Schema via dbt-duckdb

O DAG ecommerce_silver_to_gold roda dbt que lê a Silver e produz o modelo star schema. O pedido 1001 aparece na fct_orders com surrogate keys resolvidas:

🐘 Data Warehouse — Supabase

O DAG ecommerce_gold_to_dw faz TRUNCATE + INSERT das 7 tabelas Gold no schema ecommerce_analytics do Supabase. O pedido agora é acessível por ferramentas BI (Metabase, Grafana, etc.).

📊 KPI Final — Como o pedido vira número

O pedido 1001 contribui para vários KPIs via as views semânticas. Exemplo de como ele entra no KPI revenue:

-- View: v_sales_summary (schema semantic)
SELECT
    d.date_day,
    SUM(s.net_amount) AS revenue   -- Pedido 1001 contribui aqui
FROM ecommerce_analytics.fct_sales s
JOIN ecommerce_analytics.dim_date  d ON s.order_date_day = d.date_day
GROUP BY 1;

🏗️ 1. Visão Geral da Arquitetura

Componentes e Portas

🐘 2. OLTP — PostgreSQL (Fonte)

Tabela: customers

Tabela: orders

Simulação de dados por ciclo (horário)

✈️ 3. Airflow — DAGs e Orquestração

Cadeia de Triggers

DAG 1: ecommerce_data_simulator

DAG 2: ecommerce_oltp_to_raw

DAG 3: ecommerce_raw_to_bronze

Lê cada arquivo Parquet do batch, valida com Pydantic v2 e separa válidos de inválidos:

DAG 4: ecommerce_bronze_to_silver

Como o SCD2 funciona passo a passo

DAG 5: ecommerce_silver_to_gold

DAG 6: ecommerce_gold_to_dw

Carrega Gold → Supabase via TRUNCATE + INSERT em bulk (psycopg2 execute_values, page_size=1000):

📦 4. Camada Raw — MinIO Parquet

Bucket: s3://raw/ | Formato: Apache Parquet (engine pyarrow)

Particionamento: ecommerce/{table}/year=Y/month=M/day=D/hour=H/{batch_id}.parquet

Metadados adicionados em cada arquivo:

# Adicionados após extração do PostgreSQL
df["_ingested_at"]   = ingested_at        # timestamp da ingestão
df["_batch_id"]      = batch_id           # derivado do run_id do Airflow
df["_source_table"]  = "ecommerce.orders" # nome completo da tabela

🥉 5. Camada Bronze — Validação Pydantic

Bucket: s3://bronze/ | Quarantine: s3://quarantine/

Schemas Pydantic por tabela

class OrderSchema(_BaseEcommerce):
    order_id:        int
    customer_id:     int
    order_date:      datetime
    status:          str
    channel:         Optional[str]   = None
    shipping_cost:   Optional[float] = None
    discount_amount: Optional[float] = None
    created_at:      datetime
    updated_at:      datetime

    @field_validator("status")
    @classmethod
    def valid_status(cls, v):
        valid = {"pending","confirmed","shipped","delivered","cancelled","returned"}
        if str(v).lower() not in valid:
            raise ValueError(f"invalid order status: {v}")
        return v

Exemplo de registro na quarantine:

{
  "review_id": 4521,
  "rating": 6,
  "_validation_errors": "[{\"type\":\"value_error\",\"loc\":[\"rating\"],\"msg\":\"rating must be 1-5, got 6\"}]"
}

🥈 6. Camada Silver — Delta Lake

Bucket: s3://silver/ | Formato: Delta Lake (Parquet + _delta_log/)

Tabelas Silver e colunas de metadados

Código do MERGE SCD2 (Delta Lake Python):

# Fechar versão antiga (changed_pks)
df_close = df_active[df_active[pk].isin(changed_pks)].copy()
df_close["_is_current"] = False
df_close["_valid_to"]   = ts  # agora UTC

DeltaTable(silver_path, storage_options=STORAGE_OPTIONS)\
    .merge(_to_arrow(df_close), "target._sk = source._sk", "source", "target")\
    .when_matched_update_all().execute()

# Inserir nova versão
df_ins = df_new[df_new[pk].isin(to_insert)]
write_deltalake(silver_path, _to_arrow(df_ins), mode="append",
                storage_options=STORAGE_OPTIONS)

🧪 7. Qualidade de Dados — Great Expectations

Modo ephemeral — sem armazenamento persistente:

ctx   = gx.get_context(mode="ephemeral")
ds    = ctx.data_sources.add_pandas(f"ds_{table}")
asset = ds.add_dataframe_asset(f"asset_{table}")
bd    = asset.add_batch_definition_whole_dataframe("batch")
suite = ctx.suites.add(gx.ExpectationSuite(name=f"{table}_suite"))
# adiciona expectations...
result = bd.get_batch(batch_parameters={"dataframe": df}).validate(suite)
# WARN nos logs se falhar, mas retorna True (não bloqueia)

🥇 8. Camada Gold — Star Schema via dbt-duckdb

Projeto dbt: airflow/dbt/ecommerce_dbt/ | Engine: DuckDB + extensão Delta Lake

Cálculo chave: net_amount em fct_sales

-- models/gold/facts/fct_sales.sql
SELECT
    oi.order_item_id,
    oi.order_id,
    oi.product_id,
    p._sk                                                    AS product_sk,
    o.customer_id,
    c._sk                                                    AS customer_sk,
    CAST(o.order_date AS DATE)                               AS order_date,
    oi.quantity,
    oi.unit_price,
    oi.discount_pct,
    ROUND(
        oi.quantity * oi.unit_price
        * (1 - COALESCE(oi.discount_pct, 0) / 100.0), 2
    )                                                        AS net_amount
FROM delta_scan('s3://silver/ecommerce/fct_order_items') oi
LEFT JOIN delta_scan('s3://silver/ecommerce/fct_orders')    o ON o.order_id   = oi.order_id
LEFT JOIN delta_scan('s3://silver/ecommerce/dim_products')  p ON p.product_id = oi.product_id AND p._is_current = true
LEFT JOIN delta_scan('s3://silver/ecommerce/dim_customers') c ON c.customer_id= o.customer_id AND c._is_current = true

Colunas de cada tabela Gold

dim_customer

fct_sales (grain mais fino — receita por item)

dim_date (spine completa)

🐘 9. Data Warehouse — Supabase

Host: supabase-pooler:5432 | Schema analytics: ecommerce_analytics | Schema semântico: semantic

Volume estimado de dados

👁️ 10. Semantic Layer — Views Analíticas

Schema: semantic no Supabase | 9 views prontas para Metabase, Grafana, Superset

Exemplo de cohort analysis SQL:

WITH first_order AS (
    SELECT customer_id,
           DATE_TRUNC('month', MIN(order_date))::DATE AS cohort_month
    FROM ecommerce_analytics.fct_orders
    GROUP BY customer_id
)
SELECT
    f.cohort_month,
    DATE_TRUNC('month', o.order_date)::DATE               AS order_month,
    EXTRACT(MONTH FROM AGE(o.order_date, f.cohort_month))::INT AS months_since_first,
    COUNT(DISTINCT o.customer_id)                         AS active_customers,
    SUM(s.net_amount)                                     AS revenue
FROM first_order f
JOIN ecommerce_analytics.fct_orders o ON o.customer_id = f.customer_id
JOIN ecommerce_analytics.fct_sales  s ON s.order_id    = o.order_id
GROUP BY 1, 2, 3
ORDER BY 1, 3;

🔷 11. Semantic Models dbt — MetricFlow

# semantic_models.yml — trecho do modelo sales
- name: sales
  description: "Grain: order_item_id. Fonte central de receita."
  model: ref('fct_sales')
  defaults:
    agg_time_dimension: order_date

  entities:
    - name: order_item
      type: primary
      expr: order_item_id
    - name: product
      type: foreign
      expr: product_id

  measures:
    - name: revenue
      description: Receita líquida após descontos
      agg: sum
      expr: net_amount
    - name: units_sold
      agg: sum
      expr: quantity

📊 12. KPIs e Métricas — MetricFlow

Receita

Pedidos

Satisfação & Pagamentos

🔍 13. Trino — Query Engine e Time Travel

Porta: 8090 | Catálogos: gold e silver | Connector: delta_lake | Metastore: file-based

Configuração do catálogo gold:

connector.name=delta_lake
hive.metastore=file
hive.metastore.catalog.dir=s3://gold/
hive.s3.endpoint=http://minio:9000
hive.s3.path-style-access=true
hive.s3.aws-access-key=airflow
hive.s3.ssl.enabled=false
delta.enable-non-concurrent-writes=true

Exemplos de queries com time travel:

-- Versão atual
SELECT * FROM gold.ecommerce.fct_orders LIMIT 10;

-- Time travel por versão do Delta Log
SELECT * FROM gold.ecommerce.fct_orders FOR VERSION AS OF 2;

-- Time travel por timestamp
SELECT * FROM gold.ecommerce.fct_orders
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-14 00:00:00';

-- Comparar receita: versão anterior vs. atual
SELECT 'v1' AS ver, SUM(net_amount) FROM gold.ecommerce.fct_sales FOR VERSION AS OF 1
UNION ALL
SELECT 'atual',     SUM(net_amount) FROM gold.ecommerce.fct_sales;

-- Clientes ativos vs. histórico completo SCD2
SELECT 'ativos',    COUNT(*) FROM silver.ecommerce.dim_customers WHERE _is_current = true
UNION ALL
SELECT 'histórico', COUNT(*) FROM silver.ecommerce.dim_customers;

-- Receita por canal — top 3
SELECT channel, SUM(s.net_amount) AS revenue
FROM gold.ecommerce.fct_orders  o
JOIN gold.ecommerce.fct_sales   s ON s.order_id = o.order_id
GROUP BY channel
ORDER BY revenue DESC
LIMIT 3;

⚠️ 14. Problemas de Dados Intencionais

A pipeline foi projetada para exercitar cenários reais de data quality:

💧 16. Padrão Watermark — Extração Incremental

Como funciona passo a passo

Código real da pipeline

from airflow.models import Variable

# Lê watermark atual (padrão: época zero se nunca rodou)
wm_key = f"ecommerce_watermark_{table}"
last_watermark = Variable.get(wm_key, default_var="1970-01-01 00:00:00")

# Query incremental no PostgreSQL OLTP
query = f"""
    SELECT * FROM ecommerce.{table}
    WHERE updated_at > '{last_watermark}'
    ORDER BY updated_at
"""
df = pd.read_sql(query, conn)

if not df.empty:
    # Salva Parquet no MinIO
    save_to_raw(df, table)

    # Atualiza watermark para o maior updated_at do lote
    new_watermark = df["updated_at"].max().isoformat()
    Variable.set(wm_key, new_watermark)

Exceção: tabela sessions (usa created_at)

Sessões são imutáveis — uma vez criadas, nunca são atualizadas. Por isso usa created_at como watermark em vez de updated_at. Os pedidos (orders) têm ambos: created_at para novos pedidos e updated_at para mudanças de status.

Risco: late arrivals (chegadas tardias)

🚨 17. Fluxo de Quarentena — O que acontece com dados inválidos

Tipos de erro capturados

Onde ficam os dados em quarentena?

s3://quarantine/ecommerce/{table}/
  year=2026/month=03/day=16/
    quarantine_2026-03-16T15-03-22.parquet

Estrutura do registro na quarentena

Como reprocessar dados da quarentena

# 1. Ler registros com erro
import pandas as pd
df_q = pd.read_parquet("s3://quarantine/ecommerce/customers/year=2026/...")

# 2. Ver erros por tipo
import json
df_q["errors_parsed"] = df_q["_validation_errors"].apply(json.loads)
df_q["error_field"] = df_q["errors_parsed"].apply(lambda e: e[0]["field"])
print(df_q.groupby("error_field").size())

# 3. Corrigir e reenviar para bronze
df_fixed = df_q.copy()
df_fixed["email"] = df_fixed["email"].str.strip().str.lower()
# ... correções ...
df_fixed.drop(columns=["_validation_errors","_quarantined_at","_source_file"]).to_parquet(
    "s3://bronze/ecommerce/customers/reprocessed_2026-03-16.parquet"
)

🔍 18. Como Usar o Data Warehouse — Guia Prático

Queries prontas — Supabase (schema ecommerce_analytics)

Receita dos últimos 30 dias

SELECT
    d.date_day,
    ROUND(SUM(s.net_amount)::numeric, 2) AS receita_diaria,
    COUNT(DISTINCT s.order_id)           AS pedidos
FROM ecommerce_analytics.fct_sales s
JOIN ecommerce_analytics.dim_date  d ON s.order_date_day = d.date_day
WHERE d.date_day >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY 1
ORDER BY 1 DESC;

Top 10 produtos por receita

SELECT
    p.name,
    p.category,
    ROUND(SUM(s.net_amount)::numeric, 2) AS receita_total,
    SUM(s.quantity)                       AS unidades_vendidas
FROM ecommerce_analytics.fct_sales    s
JOIN ecommerce_analytics.dim_product  p ON s.product_sk = p.product_sk
GROUP BY 1, 2
ORDER BY 3 DESC
LIMIT 10;

Ticket médio e NPS por canal

SELECT
    o.channel,
    COUNT(DISTINCT o.order_id)                      AS total_pedidos,
    ROUND(AVG(o.shipping_cost)::numeric, 2)         AS frete_medio,
    ROUND(AVG(r.rating)::numeric, 2)                AS rating_medio
FROM ecommerce_analytics.fct_orders   o
LEFT JOIN ecommerce_analytics.fct_reviews r
       ON r.order_id = o.order_id
WHERE o.channel IS NOT NULL
GROUP BY 1
ORDER BY 2 DESC;

Views semânticas prontas (sem precisar escrever joins)

-- Resumo de vendas já com joins feitos
SELECT * FROM semantic.v_sales_summary    LIMIT 100;

-- Análise RFM de clientes
SELECT * FROM semantic.v_customer_rfm
WHERE recency_days < 30 AND frequency > 5
ORDER BY monetary DESC;

-- Funil de conversão
SELECT * FROM semantic.v_order_funnel;

-- Cohort de retenção
SELECT * FROM semantic.v_cohort_retention
WHERE cohort_month >= '2026-01-01';

Queries Trino — Delta Lake com Time Travel

-- Conectar ao Trino: jdbc:trino://localhost:8080/gold

-- Ver versões disponíveis de uma tabela
DESCRIBE HISTORY gold.ecommerce.fct_orders;

-- Receita ontem vs. semana passada (time travel)
SELECT 'semana_passada' AS periodo, ROUND(SUM(net_amount),2) AS receita
FROM gold.ecommerce.fct_sales
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-09 23:59:59'
UNION ALL
SELECT 'ontem', ROUND(SUM(net_amount),2)
FROM gold.ecommerce.fct_sales
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-15 23:59:59';

-- Auditoria: ver estado de um cliente em data específica
SELECT customer_id, email, address_city
FROM silver.ecommerce.dim_customers
FOR TIMESTAMP AS OF TIMESTAMP '2025-06-01 00:00:00'
WHERE customer_id = 42;

Entendendo o Star Schema — Como fazer JOINs

Checklist: o que está disponível onde

🩺 19. DAG de Monitoramento — Pipeline Health & SPC

Fluxo de execução

O que cada task verifica

Como o relatório aparece no log

╔══════════════════════════════════════════════════════════════╗
║       PIPELINE HEALTH REPORT — ecommerce — 2026-03-16       ║
╠══════════════════════════════════════════════════════════════╣
║  Overall Status: 🟢 GREEN                                    ║
╚══════════════════════════════════════════════════════════════╝

✈️  DAG RUNS             → 🟢 GREEN    (6/6 DAGs com sucesso)
📦 RAW LAYER           → 🟢 GREEN    (8/8 tables with files)
🥉 BRONZE + QUARANTINE → 🟡 YELLOW   (1 table above 5% quarantine)
🥈 SILVER DELTA        → 🟢 GREEN    (8/8 tables OK, SCD2 intact)
🥇 GOLD STAR SCHEMA    → 🟢 GREEN    (7/7 tables OK)
🐘 SUPABASE DW         → 🟢 GREEN    (7/7 counts match, 9 views OK)
📊 SPC CONTROL         → 🟢 GREEN

──── STATUS DAS DAGs DA PIPELINE ─────────────────────────────
  🟢 Data Simulator (OLTP)         [success  ]  2026-03-16 08:00  4.2 min
  🟢 OLTP → Raw                    [success  ]  2026-03-16 09:01  3.1 min
  🟢 Raw → Bronze                  [success  ]  2026-03-16 09:06  2.8 min
  🟢 Bronze → Silver               [success  ]  2026-03-16 09:10  5.4 min
  🟢 Silver → Gold                 [success  ]  2026-03-16 10:00  6.1 min
  🟢 Gold → Supabase DW            [success  ]  2026-03-16 10:08  2.4 min

──── QUARANTINE INVENTORY ────────────────────────────────────
  customers      3 rows  [email: 2, customer_id: 1]
  orders         0 rows
  products       0 rows
  reviews        7 rows  [rating: 5, review_id: 2]
  sessions       0 rows
  ...

──── SPC METRICS ─────────────────────────────────────────────
  total_orders:        today=1247.000  mean=1195.000  σ=89.000   z=+0.58  🟢
  total_revenue:       today=45230.000 mean=52100.000 σ=3120.000 z=-2.20  🟡
  quarantine_rate:     today=0.018     mean=0.021     σ=0.004    z=-0.75  🟢
  new_customers:       today=12.000    mean=11.000    σ=2.100    z=+0.47  🟢

Acionamento automático

A DAG é disparada automaticamente pelo TriggerDagRunOperator no final da ecommerce_gold_to_dw (último passo do ciclo diário). Fluxo completo:

SPC — Controle Estatístico de Processo explicado

O histórico cresce automaticamente nas primeiras 14 execuções. Com menos de 3 pontos, o SPC retorna INFO (sem dados suficientes para calcular desvio padrão).

Variáveis Airflow geradas

📖 15. Glossário Técnico