SDP 실습 — Medallion 파이프라인 구축

-- Unity Catalog에 실습용 카탈로그와 스키마 생성
CREATE CATALOG IF NOT EXISTS training;
CREATE SCHEMA IF NOT EXISTS training.ecommerce;

-- Volumes 생성
CREATE VOLUME IF NOT EXISTS training.ecommerce.raw_data;

# 샘플 주문 데이터 생성 (JSON)
import json
from datetime import datetime, timedelta
import random

orders = []
for i in range(1, 101):
    orders.append({
        "order_id": i,
        "customer_id": random.randint(1, 20),
        "product": random.choice(["노트북", "키보드", "마우스", "모니터", "헤드셋"]),
        "amount": round(random.uniform(10000, 500000), 2),
        "order_date": (datetime(2025, 1, 1) + timedelta(days=random.randint(0, 90))).isoformat(),
        "status": random.choice(["COMPLETED", "PENDING", "CANCELLED"])
    })

# JSON 파일로 저장
path = "/Volumes/training/ecommerce/raw_data/orders/"
dbutils.fs.mkdirs(path)
dbutils.fs.put(f"{path}orders_batch1.json", "\n".join(json.dumps(o) for o in orders[:50]), True)
dbutils.fs.put(f"{path}orders_batch2.json", "\n".join(json.dumps(o) for o in orders[50:]), True)

# 샘플 고객 데이터 생성 (CDC 형식)
customers = []
for i in range(1, 21):
    customers.append({
        "customer_id": i,
        "name": f"고객_{i:03d}",
        "email": f"customer{i}@example.com",
        "city": random.choice(["서울", "부산", "대구", "인천", "광주"]),
        "updated_at": datetime.now().isoformat(),
        "_change_type": "INSERT"
    })

path = "/Volumes/training/ecommerce/raw_data/customers/"
dbutils.fs.mkdirs(path)
dbutils.fs.put(f"{path}customers_initial.json", "\n".join(json.dumps(c) for c in customers), True)

-- ===== Bronze Layer =====
-- 원본 데이터를 변환 없이 그대로 수집합니다

-- 주문 데이터 수집 (Auto Loader)
CREATE OR REFRESH STREAMING TABLE bronze_orders
COMMENT 'Auto Loader로 수집한 원본 주문 데이터'
AS SELECT
    *,
    _metadata.file_path AS _source_file,     -- 원본 파일 경로
    _metadata.file_modification_time AS _file_time  -- 파일 수정 시간
FROM STREAM read_files(
    '/Volumes/training/ecommerce/raw_data/orders/',
    format => 'json'
);

-- 고객 CDC 데이터 수집
CREATE OR REFRESH STREAMING TABLE bronze_customers
COMMENT 'CDC 형식의 고객 마스터 데이터'
AS SELECT *
FROM STREAM read_files(
    '/Volumes/training/ecommerce/raw_data/customers/',
    format => 'json'
);

-- ===== Silver Layer =====
-- 데이터 타입 변환, 정제, 품질 검증을 수행합니다

-- 주문 데이터 정제 + Expectations
CREATE OR REFRESH STREAMING TABLE silver_orders (
    -- 데이터 품질 규칙 (Expectations)
    CONSTRAINT valid_order_id EXPECT (order_id IS NOT NULL)
        ON VIOLATION DROP ROW,
    CONSTRAINT valid_amount EXPECT (amount > 0)
        ON VIOLATION DROP ROW,
    CONSTRAINT valid_status EXPECT (status IN ('COMPLETED', 'PENDING', 'CANCELLED'))
        ON VIOLATION DROP ROW
)
COMMENT '정제된 주문 데이터 (품질 검증 통과)'
AS SELECT
    CAST(order_id AS BIGINT) AS order_id,
    CAST(customer_id AS BIGINT) AS customer_id,
    TRIM(product) AS product,
    CAST(amount AS DECIMAL(12,2)) AS amount,
    CAST(order_date AS TIMESTAMP) AS order_date,
    UPPER(status) AS status,
    current_timestamp() AS _processed_at    -- 처리 시각 기록
FROM STREAM(bronze_orders);

-- 고객 마스터 (CDC → SCD Type 1: 최신값 유지)
CREATE OR REFRESH STREAMING TABLE silver_customers
COMMENT '고객 마스터 데이터 (SCD Type 1)';

APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_customers)
KEYS (customer_id)
SEQUENCE BY updated_at
STORED AS SCD TYPE 1;

-- ===== Gold Layer =====
-- 비즈니스 분석에 직접 사용하는 집계 테이블입니다

-- 고객별 누적 매출
CREATE OR REFRESH MATERIALIZED VIEW gold_customer_revenue
COMMENT '고객별 매출 요약 (완료된 주문만)'
AS SELECT
    c.customer_id,
    c.name,
    c.city,
    COUNT(o.order_id) AS total_orders,
    SUM(o.amount) AS total_revenue,
    AVG(o.amount) AS avg_order_value,
    MAX(o.order_date) AS last_order_date
FROM silver_orders o
JOIN silver_customers c ON o.customer_id = c.customer_id
WHERE o.status = 'COMPLETED'
GROUP BY c.customer_id, c.name, c.city;

-- 일별 KPI
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_kpi
COMMENT '일별 주요 성과 지표'
AS SELECT
    DATE(order_date) AS sale_date,
    COUNT(DISTINCT customer_id) AS unique_customers,
    COUNT(*) AS total_orders,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value
FROM silver_orders
WHERE status = 'COMPLETED'
GROUP BY DATE(order_date);

-- Bronze: 수집된 원본 데이터 확인
SELECT COUNT(*) AS total_rows, MIN(order_date), MAX(order_date)
FROM training.ecommerce.bronze_orders;

-- Silver: 정제된 데이터 확인
SELECT status, COUNT(*) AS cnt
FROM training.ecommerce.silver_orders
GROUP BY status;

-- Silver: Expectations로 제거된 행 확인
-- (Pipeline UI의 Quality 탭에서도 확인 가능)

-- Gold: 비즈니스 집계 확인
SELECT * FROM training.ecommerce.gold_customer_revenue
ORDER BY total_revenue DESC
LIMIT 10;

SELECT * FROM training.ecommerce.gold_daily_kpi
ORDER BY sale_date;

# 추가 주문 데이터 생성 (batch3)
new_orders = []
for i in range(101, 151):
    new_orders.append({
        "order_id": i,
        "customer_id": random.randint(1, 20),
        "product": random.choice(["태블릿", "스피커", "충전기"]),
        "amount": round(random.uniform(15000, 300000), 2),
        "order_date": (datetime(2025, 4, 1) + timedelta(days=random.randint(0, 30))).isoformat(),
        "status": random.choice(["COMPLETED", "PENDING"])
    })

path = "/Volumes/training/ecommerce/raw_data/orders/"
dbutils.fs.put(f"{path}orders_batch3.json", "\n".join(json.dumps(o) for o in new_orders), True)

# 기존 고객 정보 업데이트 (도시 변경)
updates = [
    {"customer_id": 1, "name": "고객_001", "email": "customer1@example.com",
     "city": "제주", "updated_at": datetime.now().isoformat(), "_change_type": "UPDATE"},
    {"customer_id": 5, "name": "고객_005", "email": "vip5@example.com",
     "city": "서울", "updated_at": datetime.now().isoformat(), "_change_type": "UPDATE"},
]

path = "/Volumes/training/ecommerce/raw_data/customers/"
dbutils.fs.put(f"{path}customers_update1.json", "\n".join(json.dumps(u) for u in updates), True)

# CLI로 Full Refresh 실행
databricks pipelines start-update --pipeline-id <id> --full-refresh

-- 파이프라인 이벤트 로그 확인
SELECT * FROM event_log(TABLE(training.ecommerce.bronze_orders))
ORDER BY timestamp DESC
LIMIT 20;

-- Expectations 결과 상세
SELECT * FROM event_log(TABLE(training.ecommerce.silver_orders))
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC;

-- 파이프라인 삭제: Pipeline UI에서 Delete 또는
-- databricks pipelines delete --pipeline-id <id>

-- 테이블 삭제
DROP TABLE IF EXISTS training.ecommerce.bronze_orders;
DROP TABLE IF EXISTS training.ecommerce.bronze_customers;
DROP TABLE IF EXISTS training.ecommerce.silver_orders;
DROP TABLE IF EXISTS training.ecommerce.silver_customers;
DROP MATERIALIZED VIEW IF EXISTS training.ecommerce.gold_customer_revenue;
DROP MATERIALIZED VIEW IF EXISTS training.ecommerce.gold_daily_kpi;

-- 볼륨의 데이터 삭제
-- dbutils.fs.rm("/Volumes/training/ecommerce/raw_data/", True)

-- 스키마 삭제 (선택)
-- DROP SCHEMA IF EXISTS training.ecommerce CASCADE;