RisingWave

-- Kafka Source 생성 (이벤트 스트림 입력)
CREATE SOURCE user_events (
    user_id VARCHAR,
    event_type VARCHAR,
    page_url VARCHAR,
    event_time TIMESTAMPTZ,
    properties JSONB
) WITH (
    connector = 'kafka',
    topic = 'user-events',
    properties.bootstrap.server = 'redpanda:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

-- 기본 테이블 (디멘션 데이터)
CREATE TABLE users (
    user_id VARCHAR PRIMARY KEY,
    name VARCHAR,
    email VARCHAR,
    created_at TIMESTAMPTZ
);

-- Materialized View: 실시간 페이지뷰 집계
CREATE MATERIALIZED VIEW page_views_per_minute AS
SELECT
    DATE_TRUNC('minute', event_time) AS minute,
    page_url,
    COUNT(*) AS view_count,
    COUNT(DISTINCT user_id) AS unique_users
FROM user_events
WHERE event_type = 'page_view'
GROUP BY 1, 2;

-- Materialized View: 사용자 세션 분석 (윈도우 함수)
CREATE MATERIALIZED VIEW user_sessions AS
SELECT
    user_id,
    event_time,
    event_type,
    LAG(event_time) OVER (PARTITION BY user_id ORDER BY event_time) AS prev_event_time,
    CASE
        WHEN event_time - LAG(event_time) OVER (PARTITION BY user_id ORDER BY event_time) > INTERVAL '30 minutes'
        THEN 1 ELSE 0
    END AS new_session_flag
FROM user_events;

-- Materialized View: 스트림-테이블 조인
CREATE MATERIALIZED VIEW enriched_events AS
SELECT
    e.user_id,
    e.event_type,
    e.event_time,
    u.name AS user_name,
    u.email
FROM user_events e
LEFT JOIN users u ON e.user_id = u.user_id;

-- Sink: 결과를 Kafka로 출력
CREATE SINK alerts_sink FROM (
    SELECT * FROM page_views_per_minute
    WHERE view_count > 1000  -- 비정상 트래픽 감지
)
WITH (
    connector = 'kafka',
    topic = 'traffic-alerts',
    properties.bootstrap.server = 'redpanda:9092'
) FORMAT PLAIN ENCODE JSON;

-- PostgreSQL 호환 쿼리 (MV 결과 조회)
SELECT * FROM page_views_per_minute
WHERE minute >= NOW() - INTERVAL '5 minutes'
ORDER BY view_count DESC
LIMIT 10;

-- psql로 연결
-- psql -h localhost -p 4566 -d dev -U root

-- Python에서 연결 (psycopg2)
-- import psycopg2
-- conn = psycopg2.connect(
--     host="localhost", port=4566,
--     database="dev", user="root"
-- )
-- cursor = conn.cursor()
-- cursor.execute("SELECT * FROM page_views_per_minute LIMIT 10")