Medallion Architecture

Esempio: schema tabella Bronze orders

-- Bronze: raw_orders
CREATE TABLE bronze.raw_orders (
  -- Ingestion metadata
  _ingested_at      TIMESTAMP,
  _source_file      STRING,
  _batch_id         STRING,
  
  -- Raw payload (store as-is)
  raw_payload       STRING,  -- JSON string, unparsed
  
  -- Or if structured:
  order_id          STRING,  -- Not validated yet
  customer_id       STRING,
  order_date        STRING,  -- Could be any format
  amount            STRING,  -- Could have currency symbols
  status            STRING
)
PARTITIONED BY (_ingested_at::DATE)

Esempio: trasformazione Silver orders (dbt)

-- models/silver/orders.sql
WITH source AS (
  SELECT * FROM {{ source('bronze', 'raw_orders') }}
  WHERE _ingested_at >= CURRENT_DATE - INTERVAL '7 days'
),

deduplicated AS (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY order_id 
      ORDER BY _ingested_at DESC
    ) as row_num
  FROM source
),

cleaned AS (
  SELECT
    -- Cast and validate types
    CAST(order_id AS BIGINT) AS order_id,
    CAST(customer_id AS BIGINT) AS customer_id,
    
    -- Standardize date format
    TO_DATE(order_date, 'YYYY-MM-DD') AS order_date,
    
    -- Clean amount (remove currency symbols)
    CAST(REGEXP_REPLACE(amount, '[^0-9.]', '') AS DECIMAL(10,2)) AS amount,
    
    -- Normalize status to uppercase
    UPPER(TRIM(status)) AS status,
    
    -- Keep lineage
    _ingested_at,
    _batch_id
  FROM deduplicated
  WHERE row_num = 1
    AND order_id IS NOT NULL
)

SELECT * FROM cleaned

Esempio: tabella di fatti Gold (dbt)

-- models/gold/fact_orders.sql
WITH orders AS (
  SELECT * FROM {{ ref('silver_orders') }}
),

customers AS (
  SELECT * FROM {{ ref('dim_customers') }}
),

products AS (
  SELECT * FROM {{ ref('dim_products') }}
)

SELECT
  -- Surrogate key
  {{ dbt_utils.generate_surrogate_key(['o.order_id']) }} AS order_key,
  
  -- Foreign keys to dimensions
  c.customer_key,
  p.product_key,
  d.date_key,
  
  -- Degenerate dimensions
  o.order_id,
  o.status,
  
  -- Measures
  o.quantity,
  o.unit_price,
  o.quantity * o.unit_price AS line_total,
  o.discount_amount,
  
  -- Audit columns
  o._ingested_at AS source_loaded_at,
  CURRENT_TIMESTAMP AS dbt_loaded_at

FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id
LEFT JOIN products p ON o.product_id = p.product_id
LEFT JOIN {{ ref('dim_date') }} d ON o.order_date = d.date_day

# Convenzione tipica
catalog/
├── bronze/
│   ├── raw_orders
│   ├── raw_customers
│   ├── raw_products
│   └── raw_events
├── silver/
│   ├── orders           # Ordini puliti
│   ├── customers        # Clienti puliti
│   ├── products         # Prodotti puliti
│   └── events           # Eventi puliti
├── gold/
│   ├── dim_customers    # Dimensione: clienti
│   ├── dim_products     # Dimensione: prodotti
│   ├── dim_date         # Dimensione: data
│   ├── fact_orders      # Fatti: ordini
│   └── agg_daily_sales  # Aggregato: vendite giornaliere
└── quarantine/
    ├── orders_failed    # Record falliti
    └── events_failed