실습 3: SDP와 Auto Loader 통합

소스 파일
  ├── /csv/orders/       → [Bronze] bronze_orders (Streaming Table)
  └── /json/customers/   → [Bronze] bronze_customers (Streaming Table)
                               ↓
                    [Silver] silver_orders (Streaming Table, Data Quality 적용)
                    [Silver] silver_customers (Streaming Table, Data Quality 적용)
                               ↓
                    [Gold] gold_daily_sales (Materialized View)
                    [Gold] gold_customer_by_city (Materialized View)

-- =========================================================
-- [Bronze] CSV 주문 데이터 수집
-- Auto Loader(read_files)로 CSV 파일을 스트리밍 수집합니다
-- SDP가 checkpointLocation과 schemaLocation을 자동 관리합니다
-- =========================================================
CREATE OR REFRESH STREAMING TABLE bronze_orders
COMMENT '원본 주문 데이터 - CSV 파일에서 Auto Loader로 수집'
TBLPROPERTIES (
  'quality' = 'bronze',
  'pipelines.reset.allowed' = 'true'
)
AS SELECT
    *,
    _metadata.file_path          AS _source_file,
    _metadata.file_name          AS _source_file_name,
    _metadata.file_size          AS _source_file_size,
    _metadata.file_modification_time AS _file_modified_at,
    current_timestamp()          AS _ingested_at
FROM STREAM read_files(
    '/Volumes/training/auto_loader_lab/raw_data/csv/orders/',
    format            => 'csv',
    header            => true,
    inferColumnTypes  => true,
    rescuedDataColumn => '_rescued_data'
);

-- =========================================================
-- [Bronze] JSON 고객 데이터 수집
-- =========================================================
CREATE OR REFRESH STREAMING TABLE bronze_customers
COMMENT '원본 고객 데이터 - JSON 파일에서 Auto Loader로 수집'
TBLPROPERTIES (
  'quality' = 'bronze',
  'pipelines.reset.allowed' = 'true'
)
AS SELECT
    *,
    _metadata.file_path              AS _source_file,
    _metadata.file_modification_time AS _file_modified_at,
    current_timestamp()              AS _ingested_at
FROM STREAM read_files(
    '/Volumes/training/auto_loader_lab/raw_data/json/customers/',
    format            => 'json',
    inferColumnTypes  => true,
    rescuedDataColumn => '_rescued_data'
);

-- =========================================================
-- [Silver] 주문 데이터 정제 및 품질 검증
-- CONSTRAINT: 위반 행에 대해 DROP ROW 또는 경고 처리합니다
-- =========================================================
CREATE OR REFRESH STREAMING TABLE silver_orders (
    -- 데이터 품질 규칙 (Data Quality Constraints)
    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,
    CONSTRAINT no_parse_errors  EXPECT (_rescued_data IS NULL)                   ON VIOLATION DROP ROW
)
COMMENT '정제된 주문 데이터 - 품질 검증 및 타입 변환 완료'
TBLPROPERTIES ('quality' = 'silver')
AS SELECT
    CAST(order_id    AS BIGINT)        AS order_id,
    CAST(customer_id AS BIGINT)        AS customer_id,
    TRIM(product_name)                 AS product_name,
    CAST(amount      AS DECIMAL(12,2)) AS amount,
    CAST(order_date  AS DATE)          AS order_date,
    UPPER(TRIM(status))                AS status,
    _source_file,
    _ingested_at
FROM STREAM(bronze_orders)
WHERE _rescued_data IS NULL;   -- 파싱 오류 행은 Silver로 승격하지 않습니다

-- =========================================================
-- [Silver] 고객 데이터 정제
-- =========================================================
CREATE OR REFRESH STREAMING TABLE silver_customers (
    CONSTRAINT valid_customer_id EXPECT (customer_id IS NOT NULL)          ON VIOLATION DROP ROW,
    CONSTRAINT valid_email       EXPECT (email IS NOT NULL AND email LIKE '%@%') ON VIOLATION DROP ROW,
    CONSTRAINT valid_city        EXPECT (city IS NOT NULL)                 ON VIOLATION WARN  -- WARN: 경고만 기록, 행은 유지
)
COMMENT '정제된 고객 데이터'
TBLPROPERTIES ('quality' = 'silver')
AS SELECT
    CAST(customer_id AS BIGINT)       AS customer_id,
    TRIM(name)                        AS name,
    LOWER(TRIM(email))                AS email,
    TRIM(city)                        AS city,
    CAST(registered_at AS TIMESTAMP)  AS registered_at,
    -- 스키마 진화로 추가된 컬럼 (없으면 NULL)
    phone,
    membership_level,
    _source_file,
    _ingested_at
FROM STREAM(bronze_customers)
WHERE _rescued_data IS NULL;

-- =========================================================
-- [Gold] 일별 매출 요약
-- Materialized View: Silver 데이터 전체를 집계합니다
-- =========================================================
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_sales
COMMENT '일별 매출 KPI - BI 대시보드 소비용'
TBLPROPERTIES ('quality' = 'gold')
AS SELECT
    order_date,
    COUNT(*)                                                     AS order_count,
    SUM(amount)                                                  AS total_revenue,
    ROUND(AVG(amount), 2)                                        AS avg_order_value,
    COUNT(DISTINCT customer_id)                                  AS unique_customers,
    SUM(CASE WHEN status = 'COMPLETED' THEN amount ELSE 0 END)   AS completed_revenue,
    SUM(CASE WHEN status = 'CANCELLED' THEN 1     ELSE 0 END)    AS cancelled_count,
    ROUND(
        SUM(CASE WHEN status = 'COMPLETED' THEN 1 ELSE 0 END) * 100.0 / COUNT(*),
        1
    )                                                            AS completion_rate_pct
FROM silver_orders
GROUP BY order_date
ORDER BY order_date;

-- =========================================================
-- [Gold] 도시별 고객 및 매출 통계
-- =========================================================
CREATE OR REFRESH MATERIALIZED VIEW gold_customer_by_city
COMMENT '도시별 고객 및 매출 통계'
TBLPROPERTIES ('quality' = 'gold')
AS SELECT
    c.city,
    COUNT(DISTINCT c.customer_id)        AS customer_count,
    COUNT(o.order_id)                    AS order_count,
    COALESCE(SUM(o.amount), 0)           AS total_revenue,
    COALESCE(ROUND(AVG(o.amount), 2), 0) AS avg_order_value,
    -- membership_level 분포 (Silver에서 스키마 진화로 추가된 컬럼)
    COUNT(CASE WHEN c.membership_level = 'GOLD'     THEN 1 END) AS gold_members,
    COUNT(CASE WHEN c.membership_level = 'PLATINUM' THEN 1 END) AS platinum_members
FROM silver_customers c
LEFT JOIN silver_orders o ON c.customer_id = o.customer_id
GROUP BY c.city
ORDER BY total_revenue DESC;

-- Bronze 수집 확인
SELECT COUNT(*) AS bronze_orders_count FROM training.auto_loader_lab.bronze_orders;
SELECT COUNT(*) AS bronze_customers_count FROM training.auto_loader_lab.bronze_customers;

-- Silver 데이터 품질 검증
-- SDP UI Data Quality 탭에서도 확인할 수 있습니다
SELECT
    COUNT(*)                                         AS silver_orders,
    SUM(CASE WHEN order_id IS NULL THEN 1 ELSE 0 END) AS null_order_id,
    SUM(CASE WHEN amount <= 0     THEN 1 ELSE 0 END)  AS invalid_amount,
    MIN(amount) AS min_amount,
    MAX(amount) AS max_amount
FROM training.auto_loader_lab.silver_orders;

-- Gold 일별 매출 확인
SELECT
    order_date,
    order_count,
    total_revenue,
    completion_rate_pct
FROM training.auto_loader_lab.gold_daily_sales
ORDER BY order_date DESC
LIMIT 10;

-- Gold 도시별 통계
SELECT * FROM training.auto_loader_lab.gold_customer_by_city;

# 새 파일 추가
import random, json
from datetime import datetime, timedelta

new_orders = [
    f"{i},{random.randint(1000,9999)},새제품,{random.randint(10000,200000)},2025-04-01,COMPLETED"
    for i in range(201, 251)
]
dbutils.fs.put(
    "/Volumes/training/auto_loader_lab/raw_data/csv/orders/orders_batch3.csv",
    "order_id,customer_id,product_name,amount,order_date,status\n" + "\n".join(new_orders),
    overwrite=True
)
print("배치 3 생성 완료. 파이프라인을 다시 Start 하세요.")

import dlt
from pyspark.sql.functions import current_timestamp, col, upper, trim

@dlt.table(
    name="bronze_orders_py",
    comment="Python API로 정의한 Bronze 주문 테이블"
)
def bronze_orders_py():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("header", "true")
        .option("rescuedDataColumn", "_rescued_data")
        .load("/Volumes/training/auto_loader_lab/raw_data/csv/orders/")
        .withColumn("_source_file", col("_metadata.file_path"))
        .withColumn("_ingested_at", current_timestamp())
    )

@dlt.table(
    name="silver_orders_py",
    comment="Python API로 정의한 Silver 주문 테이블"
)
@dlt.expect_or_drop("valid_order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("valid_amount",   "amount > 0")
def silver_orders_py():
    return (
        dlt.read_stream("bronze_orders_py")
        .filter("_rescued_data IS NULL")
        .select(
            col("order_id").cast("bigint"),
            col("customer_id").cast("bigint"),
            trim(col("product_name")).alias("product_name"),
            col("amount").cast("decimal(12,2)"),
            col("order_date").cast("date"),
            upper(trim(col("status"))).alias("status"),
            col("_source_file"),
            col("_ingested_at")
        )
    )