Construyendo un Data Lake con Arquitectura Medallion usando DataSpoc

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│   Sources ──→ Bronze ──→ Silver ──→ Gold                    │
│   (raw)       (ingested)  (cleaned)  (business-ready)       │
│                                                             │
│   Pipe writes  Pipe writes  Lens transforms  Lens transforms│
│                                                             │
└─────────────────────────────────────────────────────────────┘

s3://company-lake/
  .dataspoc/
    manifest.json                        # Catalog (auto-updated)
    state/<pipeline>/state.json          # Incremental bookmarks
    logs/<pipeline>/<timestamp>.json     # Execution logs

  raw/                                   # ← BRONZE
    postgres/
      orders/dt=2026-04-28/orders_0000.parquet
      customers/dt=2026-04-28/customers_0000.parquet
    stripe/
      payments/dt=2026-04-28/payments_0000.parquet
    hubspot/
      contacts/dt=2026-04-28/contacts_0000.parquet

  curated/                               # ← SILVER
    finance/
      clean_orders/dt=2026-04-28/clean_orders_0000.parquet
      clean_customers/dt=2026-04-28/clean_customers_0000.parquet
    marketing/
      clean_contacts/dt=2026-04-28/clean_contacts_0000.parquet

  gold/                                  # ← GOLD
    finance/
      monthly_revenue/monthly_revenue_0000.parquet
      customer_360/customer_360_0000.parquet
    executive/
      kpi_dashboard/kpi_dashboard_0000.parquet

source:
  tap: tap-postgres
  config: ~/.dataspoc-pipe/sources/postgres-prod.json
  streams:
    - orders
    - customers
    - products

destination:
  bucket: s3://company-lake
  path: raw
  compression: zstd

incremental:
  enabled: true

schedule:
  cron: "0 */6 * * *"

{
  "host": "db.company.com",
  "port": 5432,
  "user": "dataspoc_reader",
  "dbname": "production",
  "filter_schemas": ["public"]
}

"""Transform raw Postgres data during ingestion."""

def transform(df):
    """Called per batch during extraction. Receives a pandas DataFrame."""

    # Standardize email to lowercase
    if "email" in df.columns:
        df["email"] = df["email"].str.lower().str.strip()

    # Remove test/internal records
    if "email" in df.columns:
        df = df[~df["email"].str.endswith("@test.com")]

    # Parse dates (some come as strings)
    for col in ["created_at", "updated_at"]:
        if col in df.columns:
            df[col] = pd.to_datetime(df[col], errors="coerce")

    # Drop duplicates by primary key
    if "id" in df.columns:
        df = df.drop_duplicates(subset=["id"], keep="last")

    # Remove null IDs
    if "id" in df.columns:
        df = df.dropna(subset=["id"])

    return df

source:
  tap: tap-postgres
  config: ~/.dataspoc-pipe/sources/postgres-prod.json
  streams:
    - orders
    - customers

destination:
  bucket: s3://company-lake
  path: curated/finance
  compression: zstd

incremental:
  enabled: true

schedule:
  cron: "30 */6 * * *"  # 30 min after bronze

CREATE OR REPLACE TABLE clean_orders AS
SELECT
    id,
    customer_id,
    CAST(total AS DOUBLE) AS total,
    LOWER(TRIM(status)) AS status,
    created_at,
    updated_at
FROM orders
WHERE id IS NOT NULL
  AND total > 0
  AND status IN ('pending', 'shipped', 'canceled');

CREATE OR REPLACE TABLE clean_customers AS
SELECT
    id,
    COALESCE(name, 'Unknown') AS name,
    LOWER(TRIM(email)) AS email,
    country,
    created_at
FROM customers
WHERE id IS NOT NULL
  AND email NOT LIKE '%@test.com';

CREATE OR REPLACE TABLE customer_360 AS
SELECT
    c.id AS customer_id,
    c.name,
    c.email,
    c.country,
    COUNT(o.id) AS total_orders,
    COALESCE(SUM(o.total), 0) AS lifetime_value,
    MIN(o.created_at) AS first_order,
    MAX(o.created_at) AS last_order,
    DATEDIFF('day', MAX(o.created_at), CURRENT_DATE) AS days_since_last_order,
    CASE
        WHEN DATEDIFF('day', MAX(o.created_at), CURRENT_DATE) > 90 THEN 'at_risk'
        WHEN DATEDIFF('day', MAX(o.created_at), CURRENT_DATE) > 30 THEN 'cooling'
        ELSE 'active'
    END AS status
FROM clean_customers c
LEFT JOIN clean_orders o ON c.id = o.customer_id
GROUP BY c.id, c.name, c.email, c.country;

CREATE OR REPLACE TABLE monthly_revenue AS
SELECT
    DATE_TRUNC('month', created_at) AS month,
    COUNT(*) AS order_count,
    SUM(total) AS revenue,
    COUNT(DISTINCT customer_id) AS unique_customers,
    SUM(total) / COUNT(DISTINCT customer_id) AS revenue_per_customer
FROM clean_orders
WHERE status != 'canceled'
GROUP BY 1
ORDER BY 1;

CREATE OR REPLACE TABLE kpi_dashboard AS
SELECT
    (SELECT COUNT(*) FROM clean_customers) AS total_customers,
    (SELECT COUNT(*) FROM clean_customers WHERE status = 'active') AS active_customers,
    (SELECT SUM(total) FROM clean_orders WHERE created_at >= DATE_TRUNC('month', CURRENT_DATE)) AS mtd_revenue,
    (SELECT COUNT(*) FROM clean_orders WHERE created_at >= DATE_TRUNC('month', CURRENT_DATE)) AS mtd_orders,
    (SELECT AVG(lifetime_value) FROM customer_360) AS avg_ltv,
    (SELECT COUNT(*) FROM customer_360 WHERE status = 'at_risk') AS at_risk_customers;

Every 6 hours (cron):

1. dataspoc-pipe run _ --all          # Bronze: ingest raw data
2. dataspoc-pipe run _ --all          # Silver: ingest with transforms
3. dataspoc-lens transform run        # Gold: SQL aggregations

Or automate with a simple script:

#!/bin/bash
dataspoc-pipe run postgres-prod
dataspoc-pipe run postgres-prod-clean
dataspoc-pipe run stripe-payments
dataspoc-lens transform run
echo "Medallion refresh complete at $(date)"

-- Bronze: raw data (debug, audit)
lens> SELECT * FROM orders LIMIT 5;

-- Silver: clean data (analysis)
lens> SELECT * FROM clean_orders WHERE status = 'shipped' LIMIT 5;

-- Gold: business metrics (dashboards, reports)
lens> SELECT * FROM monthly_revenue ORDER BY month DESC LIMIT 12;
lens> SELECT * FROM customer_360 WHERE status = 'at_risk';
lens> SELECT * FROM kpi_dashboard;

User: "Give me a summary of this month's KPIs."

Agent: [MCP] query("SELECT * FROM kpi_dashboard")

Agent: "Here's this month's performance:
  - 12,847 total customers (9,231 active)
  - $487k MTD revenue from 3,241 orders
  - Average LTV: $1,247
  - 847 customers flagged as at-risk (no order in 90+ days)"