3.02 · Deep Dive: Silver Layer - Data Cleaning & Validation

graph LR
    B["🥉 Bronze<br/>(Raw, As-Is)"]
    SC["Data Quality<br/>Checks"]
    DD["Deduplication<br/>Logic"]
    TR["Type<br/>Conversions"]
    JN["Join &<br/>Enrich"]
    S["🥈 Silver<br/>(Validated,<br/>Deduplicated)"]

    B -->|Detect anomalies| SC
    SC -->|Remove duplicates| DD
    DD -->|Cast to business types| TR
    TR -->|Join reference data| JN
    JN -->|Load| S

-- Silver layer: Validated customers table
CREATE TABLE silver.customers (
  customer_id INT,
  email STRING,
  phone STRING,
  created_at TIMESTAMP,
  updated_at TIMESTAMP,
  _load_date DATE,
  _is_valid BOOLEAN,
  _quality_issues ARRAY<STRING>
);

-- Validation logic
INSERT INTO silver.customers
WITH validated AS (
  SELECT
    customer_id,
    email,
    phone,
    created_at,
    updated_at,
    CURRENT_DATE() as _load_date,
    CASE
      WHEN customer_id IS NULL THEN FALSE
      WHEN email NOT LIKE '%@%.%' THEN FALSE
      WHEN phone NOT LIKE '+%' AND phone IS NOT NULL THEN FALSE
      WHEN created_at > CURRENT_TIMESTAMP() THEN FALSE
      ELSE TRUE
    END as _is_valid,
    CASE
      WHEN customer_id IS NULL THEN ['missing_customer_id']
      WHEN email NOT LIKE '%@%.%' THEN ['invalid_email']
      WHEN created_at > CURRENT_TIMESTAMP() THEN ['future_created_date']
      ELSE []
    END as _quality_issues
  FROM bronze.customers
)
SELECT * FROM validated;

-- Deduplication using ROW_NUMBER()
WITH deduped AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY customer_id, email
      ORDER BY _elt_timestamp DESC
    ) as rn
  FROM bronze.customers
)
SELECT * EXCEPT (rn)
FROM deduped
WHERE rn = 1;  -- Keep only the latest version

-- When the same customer appears twice in one load
-- (e.g., bug in source system produces duplicate rows)

WITH deduped AS (
  SELECT
    customer_id,
    email,
    phone,
    -- Keep the row with most recent update
    CASE 
      WHEN ROW_NUMBER() OVER (
        PARTITION BY customer_id 
        ORDER BY updated_at DESC, _elt_timestamp DESC
      ) = 1 THEN TRUE 
      ELSE FALSE 
    END as is_current
  FROM bronze.customers
)
SELECT *
FROM deduped
WHERE is_current = TRUE;

-- Convert to business-friendly types
CREATE OR REPLACE TABLE silver.customers AS
SELECT
  CAST(customer_id AS INT64) as customer_id,
  LOWER(TRIM(email)) as email,  -- Normalize email (lowercase, trim)
  REGEXP_REPLACE(phone, '[^0-9+]', '') as phone,  -- Remove formatting
  CAST(created_at AS TIMESTAMP) as created_at,
  CAST(updated_at AS TIMESTAMP) as updated_at,

  -- Add technical metadata
  CURRENT_DATE() as _load_date,
  1 as _dbt_version  -- Track transformation version
FROM bronze.customers
WHERE _is_valid = TRUE;  -- Only keep validated rows

-- Silver layer: Add business context from dimension tables
CREATE OR REPLACE TABLE silver.customers_enriched AS
SELECT
  c.customer_id,
  c.email,
  c.phone,
  c.created_at,

  -- Enrich with geography reference
  g.country_name,
  g.region_id,

  -- Enrich with segment reference
  s.customer_segment,
  s.lifetime_value_bucket,

  c._load_date
FROM silver.customers c
LEFT JOIN silver.dim_geography g
  ON c.country_code = g.country_code
LEFT JOIN silver.customer_segments s
  ON c.customer_id = s.customer_id
WHERE c._load_date = CURRENT_DATE();

-- SCD Type 6: Keep all versions with effective dates
CREATE TABLE silver.customers_scd (
  customer_key INT64,  -- Surrogate key
  customer_id INT64,   -- Business key
  email STRING,
  phone STRING,

  -- SCD Type 2 Tracking
  effective_date DATE,
  end_date DATE,
  is_current BOOLEAN,

  _load_date DATE,
  _version INT
);

-- Insert new versions
INSERT INTO silver.customers_scd
WITH latest_bronze AS (
  SELECT DISTINCT
    customer_id,
    email,
    phone,
    CURRENT_DATE() as effective_date,
    NULL as end_date,
    TRUE as is_current
  FROM bronze.customers
  WHERE _is_valid = TRUE
)
SELECT
  GENERATE_UNIQUE_ID() as customer_key,
  customer_id,
  email,
  phone,
  effective_date,
  end_date,
  is_current,
  CURRENT_DATE() as _load_date,
  1 as _version
FROM latest_bronze;

-- MERGE: Update existing, insert new
MERGE INTO silver.customers_scd t
USING (
  SELECT DISTINCT
    customer_id,
    email,
    phone,
    CURRENT_DATE() as effective_date
  FROM bronze.customers
  WHERE _is_valid = TRUE
) s
ON t.customer_id = s.customer_id AND t.is_current = TRUE
WHEN MATCHED AND t.email != s.email THEN
  UPDATE SET
    end_date = CURRENT_DATE() - 1,
    is_current = FALSE
WHEN NOT MATCHED THEN
  INSERT (customer_key, customer_id, email, phone, effective_date, end_date, is_current, _load_date)
  VALUES (GENERATE_UNIQUE_ID(), s.customer_id, s.email, s.phone, s.effective_date, NULL, TRUE, CURRENT_DATE());

# dbt/models/silver/silver_customers.yml
version: 2

models:
  - name: silver_customers
    description: "Validated and deduplicated customer dimension"
    columns:
      - name: customer_id
        tests:
          - not_null
          - unique
          - relationships:
              to: ref('bronze_customers')
              field: customer_id

      - name: email
        tests:
          - not_null
          - custom_regex:
              pattern: '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'

      - name: created_at
        tests:
          - not_null
          - custom:
              expression: '{{ column_name }} < current_timestamp()'

bronze.customers
    ├─→ silver.customers (SCD Type 2)
    ├─→ silver.customer_emails (deduplicated)
    └─→ silver.customer_addresses (deduplicated, joined with reference)

WITH scored_dupes AS (
  SELECT
    *,
    CASE
      WHEN _data_quality_score > 0.9 THEN 1  -- High quality = keep first
      ELSE 2  -- Lower quality = keep latest update
    END as priority
  FROM bronze.customers
)
SELECT *
FROM (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY customer_id
      ORDER BY priority ASC, updated_at DESC
    ) as rn
  FROM scored_dupes
)
WHERE rn = 1;