Engenharia de Dados

  ┌─────────────────────────────────────────────────────────────────┐
  │                       FONTES DE DADOS                            │
  │  Postgres │ Salesforce │ Stripe │ Eventos (Kafka) │ APIs SaaS    │
  └────────────────────────────┬────────────────────────────────────┘
                               │
                ┌──────────────▼──────────────┐
                │   INGESTION (Fivetran/      │
                │   Airbyte/Kafka Connect)    │   ← ELT, CDC, streaming
                └──────────────┬──────────────┘
                               │
                ┌──────────────▼──────────────┐
                │   RAW LAYER (Bronze)        │   ← snapshot bruto
                │   Snowflake / BigQuery /    │     immutable, append-only
                │   Databricks / S3 + Iceberg │
                └──────────────┬──────────────┘
                               │
                ┌──────────────▼──────────────┐
                │   TRANSFORMATION (dbt)      │   ← Silver: limpo, tipado
                │   - staging                 │     Gold: dim/fact, metrics
                │   - intermediate            │
                │   - marts (gold)            │
                └──────────────┬──────────────┘
                               │
        ┌──────────────────────┼──────────────────────────┐
        │                      │                          │
        ▼                      ▼                          ▼
  ┌──────────┐         ┌─────────────┐           ┌────────────────┐
  │   BI     │         │ Reverse ETL │           │  AI / ML Layer │
  │ Looker   │         │  Hightouch  │           │ Vector DB      │
  │ Tableau  │         │  → Salesforce│          │ Feature Store  │
  │ Metabase │         │  → HubSpot  │           │ LLM (RAG)      │
  └──────────┘         └─────────────┘           └────────────────┘
        ▲                                                ▲
        │                                                │
   ORQUESTRAÇÃO (Airflow / Dagster / Prefect) ───────────┘
   OBSERVABILIDADE (Monte Carlo / Soda / dbt tests)
   GOVERNANÇA (Atlan / Collibra / DataHub)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp

# ─── Spark configurado para Iceberg em S3 ─────────────────────────────────────
# Em produção, essas configs ficam em spark-defaults.conf ou no cluster manager
spark = (
    SparkSession.builder
    .appName("iceberg-demo")
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.lakehouse.type", "glue")  # ou "hive", "rest", "nessie"
    .config("spark.sql.catalog.lakehouse.warehouse", "s3://meu-lakehouse/warehouse/")
    .getOrCreate()
)

# ─── 1. Criando uma tabela Iceberg com partition evolution ────────────────────
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.vendas.pedidos (
        pedido_id   BIGINT,
        cliente_id  BIGINT,
        valor       DECIMAL(10,2),
        status      STRING,
        criado_em   TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(criado_em))   -- hidden partitioning: única do Iceberg
    TBLPROPERTIES (
        'format-version' = '2',         -- v2 suporta row-level deletes (MERGE)
        'write.target-file-size-bytes' = '134217728'  -- 128MB sweet spot
    )
""")

# ─── 2. Insert + MERGE (upsert) ───────────────────────────────────────────────
novos_pedidos = spark.read.parquet("s3://landing/pedidos-2026-04-22/")

# MERGE = padrão de upsert ACID que antes só existia em DBs
spark.sql("""
    MERGE INTO lakehouse.vendas.pedidos AS alvo
    USING (SELECT * FROM novos_pedidos_view) AS origem
    ON alvo.pedido_id = origem.pedido_id
    WHEN MATCHED AND origem.status != alvo.status THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

# ─── 3. Time Travel: lendo o estado da tabela em qualquer ponto do passado ───
# Por timestamp (auditoria, debugging de dados, "ontem o número estava certo")
df_ontem = (
    spark.read
    .option("as-of-timestamp", "2026-04-21 23:59:59")
    .table("lakehouse.vendas.pedidos")
)
print(f"Pedidos ontem: {df_ontem.count():,}")

# Por snapshot_id (reprodutibilidade exata de um experimento de ML)
snapshots = spark.sql("SELECT * FROM lakehouse.vendas.pedidos.snapshots").collect()
snapshot_treino = snapshots[0].snapshot_id
df_treino_v1 = (
    spark.read
    .option("snapshot-id", snapshot_treino)
    .table("lakehouse.vendas.pedidos")
)

# ─── 4. Schema evolution sem rewrite ──────────────────────────────────────────
# Isso é o que faz Iceberg brilhar: você adiciona coluna em PB de dados
# em milissegundos (só metadado muda, arquivos Parquet ficam intactos).
spark.sql("""
    ALTER TABLE lakehouse.vendas.pedidos
    ADD COLUMN canal STRING COMMENT 'Origem do pedido (web/app/loja)'
""")

# Renomear coluna sem reescrever nada
spark.sql("ALTER TABLE lakehouse.vendas.pedidos RENAME COLUMN valor TO valor_brl")

# ─── 5. Manutenção: compaction de small files ─────────────────────────────────
# Streaming gera muitos arquivos pequenos. Compaction agrupa em arquivos do
# tamanho ideal para query performance.
spark.sql("""
    CALL lakehouse.system.rewrite_data_files(
        table => 'vendas.pedidos',
        options => map('target-file-size-bytes', '134217728')
    )
""")

# Expira snapshots antigos (>7 dias) para liberar storage
spark.sql("""
    CALL lakehouse.system.expire_snapshots(
        table => 'vendas.pedidos',
        older_than => TIMESTAMP '2026-04-15 00:00:00'
    )
""")

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
import time
from datetime import datetime

# ─── PRODUCER: emite eventos de pedidos para o tópico "pedidos.criados" ──────
producer = KafkaProducer(
    bootstrap_servers=["broker-1:9092", "broker-2:9092", "broker-3:9092"],
    # Serialização: dict Python → JSON UTF-8
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    # Chave usada para particionamento: pedidos do mesmo cliente vão na mesma
    # partição, garantindo ordenação por cliente.
    key_serializer=lambda k: str(k).encode("utf-8"),
    acks="all",                # espera confirmação de todas as réplicas (durabilidade)
    enable_idempotence=True,   # evita duplicatas em caso de retry
    retries=10,
    linger_ms=5,               # buffer 5ms para batching mais eficiente
    compression_type="snappy"  # compressão reduz custo de rede ~70%
)

def publicar_pedido(pedido: dict):
    """Publica evento de pedido no tópico, particionado por cliente_id."""
    try:
        future = producer.send(
            topic="pedidos.criados",
            key=pedido["cliente_id"],
            value={**pedido, "publicado_em": datetime.utcnow().isoformat()}
        )
        # .get() bloqueia até confirmação — em prod, prefira callback assíncrono
        metadata = future.get(timeout=10)
        print(f"✓ partição={metadata.partition} offset={metadata.offset}")
    except KafkaError as e:
        print(f"✗ Falha ao publicar: {e}")

# Exemplo de uso
publicar_pedido({"pedido_id": 12345, "cliente_id": 789, "valor_brl": 199.90})
producer.flush()  # garante que tudo foi enviado antes de encerrar

# ─── CONSUMER: detecta fraude em tempo real ──────────────────────────────────
consumer = KafkaConsumer(
    "pedidos.criados",
    bootstrap_servers=["broker-1:9092"],
    group_id="deteccao-fraude-v1",     # consumer group: paralelismo + offset tracking
    auto_offset_reset="earliest",       # do início se não houver offset salvo
    enable_auto_commit=False,           # commit manual = at-least-once explícito
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    max_poll_records=100,               # batch size por poll
    session_timeout_ms=30000
)

def avaliar_fraude(pedido: dict) -> bool:
    """Heurística simplificada — em produção seria modelo ML em feature store."""
    return pedido["valor_brl"] > 10_000 or pedido.get("ip_pais") == "XX"

print("🎧 Aguardando eventos...")
for mensagem in consumer:
    pedido = mensagem.value
    if avaliar_fraude(pedido):
        # Emite evento de alerta em outro tópico — composição de streams
        producer.send("alertas.fraude", key=pedido["cliente_id"], value=pedido)
        print(f"🚨 Fraude suspeita: pedido {pedido['pedido_id']}")
    # Commit manual só após processamento bem-sucedido (at-least-once correto)
    consumer.commit()

import great_expectations as gx
import pandas as pd
from datetime import datetime

# ─── Setup do contexto GE (em prod, persistido em S3/Git) ────────────────────
context = gx.get_context(mode="ephemeral")  # ephemeral = só na memória, p/ demo

# ─── Carregando dados que queremos validar ───────────────────────────────────
df = pd.read_parquet("s3://lakehouse/silver/clientes/dt=2026-04-22/")

# Conecta o DataFrame ao contexto GE
data_source = context.data_sources.add_pandas("clientes_silver")
data_asset = data_source.add_dataframe_asset(name="clientes_diario")
batch_definition = data_asset.add_batch_definition_whole_dataframe("batch_full")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})

# ─── Definindo as expectations (o "contrato" da tabela) ──────────────────────
suite = context.suites.add(gx.ExpectationSuite(name="clientes_silver_v1"))

# 1. Schema esperado: colunas obrigatórias
suite.add_expectation(gx.expectations.ExpectTableColumnsToMatchSet(
    column_set=["cliente_id", "email", "cpf", "criado_em", "status", "score_credito"],
    exact_match=False  # permite colunas extras, exige as listadas
))

# 2. Chave primária: cliente_id deve ser único e nunca nulo
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="cliente_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="cliente_id"))

# 3. Email: formato válido (regex). Campos de PII como email JAMAIS podem ter
#    valores quebrados que indiquem corrupção upstream.
suite.add_expectation(gx.expectations.ExpectColumnValuesToMatchRegex(
    column="email",
    regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
    mostly=0.99  # tolera 1% de inválidos (legado), alerta se passar disso
))

# 4. CPF: comprimento e formato (não validamos dígito aqui, mas poderíamos)
suite.add_expectation(gx.expectations.ExpectColumnValueLengthsToEqual(
    column="cpf", value=11
))

# 5. Score de crédito: regra de negócio — sempre entre 0 e 1000
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="score_credito", min_value=0, max_value=1000
))

# 6. Frescor: criado_em deve estar dentro das últimas 24h (SLA do contrato)
suite.add_expectation(gx.expectations.ExpectColumnMaxToBeBetween(
    column="criado_em",
    min_value=(datetime.utcnow() - pd.Timedelta(hours=24)).isoformat(),
    max_value=datetime.utcnow().isoformat()
))

# 7. Status: enum fechado — qualquer valor fora disso quebra contrato
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(
    column="status",
    value_set=["ATIVO", "INATIVO", "BLOQUEADO", "PENDENTE"]
))

# ─── Executando validação ────────────────────────────────────────────────────
validation_definition = context.validation_definitions.add(
    gx.ValidationDefinition(
        name="clientes_diario_validacao",
        data=batch_definition,
        suite=suite
    )
)
result = validation_definition.run(batch_parameters={"dataframe": df})

# ─── Reagindo ao resultado ───────────────────────────────────────────────────
if not result.success:
    falhas = [r for r in result.results if not r.success]
    print(f"❌ {len(falhas)} expectations falharam — pipeline ABORTADO")
    for f in falhas:
        print(f"  - {f.expectation_config.type}: {f.result.get('unexpected_count', 'N/A')} valores ruins")
    # Em produção: chamar Airflow/Dagster para sair com erro, alertar Slack,
    # bloquear promoção de Silver → Gold.
    raise RuntimeError("Quality gate failed")
else:
    print(f"✅ {len(result.results)} expectations passaram — promovendo para Gold")

import openai
import os
import sqlglot                       # parser SQL multi-dialeto, p/ validação
from sqlglot import expressions as exp

client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# ─── 1. Schema curado: o "contexto" que o LLM recebe ─────────────────────────
# Em produção, isso vem do catálogo (DataHub/Atlan) ou de uma semantic layer.
SCHEMA_DOC = """
-- Tabela: analytics.gold.fct_pedidos
-- Granularidade: 1 linha por pedido. Atualizada a cada 15 min.
CREATE TABLE analytics.gold.fct_pedidos (
    pedido_id      BIGINT       PRIMARY KEY,
    cliente_id     BIGINT       REFERENCES dim_clientes,
    produto_id     BIGINT       REFERENCES dim_produtos,
    valor_brl      DECIMAL(10,2),  -- Valor líquido em BRL
    canal          STRING,       -- 'web' | 'app' | 'loja'
    criado_em      TIMESTAMP,    -- UTC
    status         STRING        -- 'PAGO' | 'CANCELADO' | 'PENDENTE'
);

-- Tabela: analytics.gold.dim_clientes
-- Granularidade: 1 linha por cliente. SCD tipo 2.
CREATE TABLE analytics.gold.dim_clientes (
    cliente_id     BIGINT,
    nome           STRING,
    estado_uf      STRING,       -- 'SP', 'RJ', 'RS', etc.
    regiao         STRING,       -- 'Sul' | 'Sudeste' | 'Norte' | ...
    segmento       STRING        -- 'B2B' | 'B2C'
);
"""

SYSTEM_PROMPT = f"""Você é um analista SQL sênior em Snowflake.
Gere APENAS uma query SQL válida em Snowflake SQL, nada mais.

Regras OBRIGATÓRIAS:
- Use apenas as tabelas e colunas listadas no schema abaixo.
- SEMPRE inclua LIMIT 1000 a menos que o usuário peça agregação.
- Use joins explícitos (INNER/LEFT JOIN), nunca cross join implícito.
- NUNCA gere DDL, DML (UPDATE/DELETE/INSERT), nem queries em outras tabelas.
- Para datas relativas, use CURRENT_DATE() / DATEADD().

Schema disponível:
{SCHEMA_DOC}
"""

def texto_para_sql(pergunta: str) -> str:
    """Converte pergunta em PT para SQL Snowflake."""
    resp = client.chat.completions.create(
        model="gpt-4o-mini",          # bom o bastante p/ Text-to-SQL com schema
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": pergunta}
        ],
        temperature=0.0,              # determinístico — SQL não é criativo
    )
    sql = resp.choices[0].message.content.strip()
    # Remove ```sql blocos se o modelo embrulhar
    sql = sql.removeprefix("```sql").removeprefix("```").removesuffix("```").strip()
    return sql

def validar_sql_seguro(sql: str) -> tuple[bool, str]:
    """Guardrail: parseia o SQL e rejeita operações perigosas."""
    try:
        parsed = sqlglot.parse_one(sql, read="snowflake")
    except Exception as e:
        return False, f"SQL inválido: {e}"

    # Bloqueia qualquer DDL/DML — LLM só pode SELECT
    proibidos = (exp.Insert, exp.Update, exp.Delete, exp.Drop, exp.AlterTable, exp.Create)
    if any(parsed.find(p) for p in proibidos):
        return False, "Operação proibida detectada (DDL/DML)"

    # Tabelas referenciadas devem estar na allowlist
    ALLOWLIST = {"analytics.gold.fct_pedidos", "analytics.gold.dim_clientes"}
    tabelas = {t.name.lower() for t in parsed.find_all(exp.Table)}
    if not tabelas.issubset({n.split(".")[-1] for n in ALLOWLIST}):
        return False, f"Tabela não permitida: {tabelas - ALLOWLIST}"

    return True, "OK"

# ─── Uso ──────────────────────────────────────────────────────────────────────
pergunta = "Quanto vendemos para clientes do Sul no último trimestre, por canal?"
sql_gerado = texto_para_sql(pergunta)
print(f"📝 SQL gerado:\n{sql_gerado}\n")

ok, msg = validar_sql_seguro(sql_gerado)
if ok:
    print("✅ Guardrail aprovou — pode executar no warehouse")
    # cursor.execute(sql_gerado) ...
else:
    print(f"🚫 Bloqueado: {msg}")