Streaming Tables & Materialized Views

-- 기본 Streaming Table: 새 데이터를 증분 처리
CREATE OR REFRESH STREAMING TABLE silver_orders
AS SELECT
    CAST(order_id AS BIGINT) AS order_id,
    CAST(customer_id AS BIGINT) AS customer_id,
    CAST(amount AS DECIMAL(12,2)) AS amount,
    CAST(order_date AS TIMESTAMP) AS order_date,
    CAST(status AS STRING) AS status
FROM STREAM(bronze_orders);

import dlt
from pyspark.sql.functions import col

@dlt.table(
    comment="정제된 주문 데이터 (증분 처리)",
    table_properties={"quality": "silver"}
)
def silver_orders():
    return (
        dlt.read_stream("bronze_orders")
           .select(
               col("order_id").cast("bigint"),
               col("customer_id").cast("bigint"),
               col("amount").cast("decimal(12,2)"),
               col("order_date").cast("timestamp"),
               col("status").cast("string")
           )
    )

-- 일별 매출 집계 (Materialized View)
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_revenue
AS SELECT
    DATE(order_date) AS sale_date,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value,
    COUNT(DISTINCT customer_id) AS unique_customers
FROM silver_orders
GROUP BY DATE(order_date);

import dlt
from pyspark.sql.functions import col, count, sum as _sum, avg, countDistinct, date_trunc

@dlt.table(
    comment="일별 매출 집계",
    table_properties={"quality": "gold"}
)
def gold_daily_revenue():
    return (
        dlt.read("silver_orders")  # STREAM이 아닌 일반 read
           .groupBy(date_trunc("day", col("order_date")).alias("sale_date"))
           .agg(
               count("*").alias("order_count"),
               _sum("amount").alias("total_revenue"),
               avg("amount").alias("avg_order_value"),
               countDistinct("customer_id").alias("unique_customers")
           )
    )

-- ============================================
-- Bronze: Streaming Table (원본 데이터 수집)
-- ============================================
CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
    '/volumes/catalog/schema/landing/orders/',
    format => 'json'
);

CREATE OR REFRESH STREAMING TABLE bronze_customers
AS SELECT * FROM STREAM read_files(
    '/volumes/catalog/schema/landing/customers/',
    format => 'json'
);

-- ============================================
-- Silver: Streaming Table (데이터 정제)
-- ============================================
CREATE OR REFRESH STREAMING TABLE silver_orders (
    CONSTRAINT valid_order_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT positive_amount EXPECT (amount > 0) ON VIOLATION DROP ROW
)
AS SELECT
    CAST(order_id AS BIGINT) AS order_id,
    CAST(customer_id AS BIGINT) AS customer_id,
    CAST(amount AS DECIMAL(12,2)) AS amount,
    CAST(order_date AS TIMESTAMP) AS order_date,
    CAST(status AS STRING) AS status
FROM STREAM(bronze_orders);

-- Silver: CDC 처리
CREATE OR REFRESH STREAMING TABLE silver_customers;

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

-- ============================================
-- Gold: Materialized View (비즈니스 집계)
-- ============================================
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_revenue
AS SELECT
    DATE(o.order_date) AS sale_date,
    COUNT(*) AS order_count,
    SUM(o.amount) AS total_revenue,
    AVG(o.amount) AS avg_order_value
FROM silver_orders o
GROUP BY DATE(o.order_date);

CREATE OR REFRESH MATERIALIZED VIEW gold_customer_orders
AS SELECT
    c.customer_id,
    c.name AS customer_name,
    c.city,
    COUNT(o.order_id) AS total_orders,
    SUM(o.amount) AS total_spent,
    MAX(o.order_date) AS last_order_date
FROM silver_customers c
LEFT JOIN silver_orders o ON c.customer_id = o.customer_id
GROUP BY c.customer_id, c.name, c.city;

-- SQL에서 전체 새로고침 (소스부터 재처리)
CREATE OR REFRESH STREAMING TABLE silver_orders
AS SELECT ...
FROM STREAM(bronze_orders);
-- Pipeline UI에서 "Full Refresh" 옵션 선택