5. Confluent Cloud for Apache Flink Lab 2 따라하기

-- 고객 ID가 3001인 주문에 대해 고객 이메일 정보를 조인
SELECT 
    order_id, 
    unique_orders.`$rowtime`,   -- 주문 발생 시각 (이벤트 시간)
    email                       -- 고객 이메일 정보
FROM unique_orders
INNER JOIN customers           -- 일반 조인 (Regular Join)
ON unique_orders.customer_id = customers.customer_id
WHERE unique_orders.customer_id = 3001;

-- 고객 ID가 3001인 주문에 대해 주문 시점의 고객 이메일 정보를 조회
SELECT 
    order_id, 
    unique_orders.`$rowtime`,   -- 주문 발생 시각 (이벤트 시간)
    email                       -- 주문 당시의 고객 이메일 (변경 불가)
FROM unique_orders
-- Temporal Join을 사용하여 주문 발생 시점 기준으로 고객 정보 조회
INNER JOIN customers FOR SYSTEM_TIME AS OF unique_orders.`$rowtime`
ON unique_orders.customer_id = customers.customer_id
WHERE unique_orders.customer_id = 3001;

-- 상태 관리 설정
SET 'client.statement-name' = 'valid-orders-materializer';

-- 유효 주문 테이블 생성
-- 주문과 결제를 시간 조건으로 조인하여 유효한 주문 추출
-- 결제 시각 기준으로 10분 이내에 발생한 주문만 매칭
CREATE TABLE valid_orders (
    order_id STRING,
    customer_id INT,
    product_id STRING,
    order_time TIMESTAMP_LTZ(3),
    payment_time TIMESTAMP_LTZ(3),
    amount DECIMAL,
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND  -- 이벤트 시간 기준 워터마크 설정
) AS 
SELECT 
    unique_orders.order_id,
    customer_id,
    product_id,
    unique_orders.`$rowtime` AS order_time,  -- 주문 시각
    payment_time,                            -- 결제 시각
    amount                                   -- 결제 금액
FROM unique_orders
INNER JOIN payments
ON unique_orders.order_id = payments.order_id
-- 주문 시각이 결제 시각보다 10분 전 ~ 결제 시각 사이인 경우만 유지
WHERE unique_orders.`$rowtime` 
  BETWEEN payment_time - INTERVAL '10' MINUTES AND payment_time;

-- 복합 데이터 프로덕트 설정
SET 'client.statement-name' = 'valid-orders-customer-product-materializer';

-- 주문 + 고객 + 상품 정보를 통합한 데이터 프로덕트 생성
CREATE TABLE order_customer_product (
    order_id STRING,
    customer_id INT,
    name STRING,
    email STRING,
    brand STRING,
    product STRING,
    sale_price DOUBLE
) WITH ('changelog.mode' = 'retract')  -- 변경 사항 추적을 위한 CDC 설정
AS 
SELECT 
    valid_orders.order_id,
    valid_orders.customer_id,
    customers.name,
    customers.email,
    products.brand,
    products.name AS product,
    valid_orders.amount AS sale_price
FROM valid_orders
-- 주문 시점 기준 고객 정보 조인 (Temporal Join)
INNER JOIN customers FOR SYSTEM_TIME AS OF valid_orders.order_time
    ON valid_orders.customer_id = customers.customer_id
-- 주문 시점 기준 상품 정보 조인 (Temporal Join)
INNER JOIN products FOR SYSTEM_TIME AS OF valid_orders.order_time
    ON valid_orders.product_id = products.product_id;

-- 전자제품 프로모션 대상 고객 식별
-- 특정 브랜드 제품을 많이 구매한 고객을 프로모션 대상으로 추출
SELECT 
    customer_id,
    COLLECT(brand) AS products,        -- 구매한 브랜드 목록 수집
    'bundle_offer' AS promotion_name   -- 고정 프로모션 이름
FROM order_customer_product
WHERE brand IN ('Samsung', 'Sony', 'LG')  -- 대상 브랜드 필터
GROUP BY customer_id
HAVING COUNT(DISTINCT brand) >= 2         -- 서로 다른 브랜드 2개 이상
   AND COUNT(brand) > 5;                  -- 총 5개 이상 구매

-- 프로모션 테이블 생성
CREATE TABLE electronics_promotions (
    customer_id INT,
    promotion_name STRING,
    PRIMARY KEY (customer_id) NOT ENFORCED
);

-- 프로모션 데이터 삽입
SET 'client.statement-name' = 'electronics-promotions-materializer';

INSERT INTO electronics_promotions
SELECT 
    customer_id,
    'bundle_offer' AS promotion_name
FROM order_customer_product
WHERE brand IN ('Samsung', 'Sony', 'LG')
GROUP BY customer_id
HAVING COUNT(DISTINCT brand) >= 2 AND COUNT(brand) > 5;

-- 고객별 시간대별 주문 통계
-- 고객별 최근 1시간 내 주문 수와 총 결제 금액 집계
SELECT 
    customer_id,
    order_id,
    order_time,
    COUNT(*) OVER w AS num_orders,   -- 최근 1시간 주문 수
    SUM(amount) OVER w AS total_price -- 최근 1시간 총 결제 금액
FROM `valid_orders`
WINDOW w AS (
    PARTITION BY customer_id           -- 고객별로 집계 분리
    ORDER BY order_time                -- 주문 시각 기준 정렬
    RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW -- 최근 1시간 범위
);

-- CTE를 활용한 고객 등급 분류
CREATE TABLE reward_levels (
    customer_id BIGINT, 
    total_price DOUBLE,
    rewards_level STRING,
    updated_at TIMESTAMP_LTZ(3),
    PRIMARY KEY (customer_id) NOT ENFORCED
) AS 
WITH total_price_per_customer_1h AS (
    -- 고객별 최근 1시간 누적 결제 금액
    SELECT 
        customer_id,
        SUM(amount) OVER w AS total_price, 
        order_time
    FROM `valid_orders`
    WINDOW w AS (
        PARTITION BY customer_id
        ORDER BY order_time
        RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
    )
) 
-- 금액 구간별 등급 부여
SELECT 
    COALESCE(customer_id, 0) AS customer_id,     -- null 방지
    total_price,
    CASE
        WHEN total_price > 30000 THEN 'GOLD'
        WHEN total_price > 20000 THEN 'SILVER'
        WHEN total_price > 10000 THEN 'BRONZE'
        ELSE 'NONE'
    END AS rewards_level,
    order_time AS updated_at
FROM total_price_per_customer_1h;