Kafka 이벤트를 서브초 쿼리로 — ClickHouse Materialized View·Projections로 초당 수만 건 집계 파이프라인 설계하기

CREATE TABLE events_kafka
(
    event_id    String,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    country     LowCardinality(String),
    ts          DateTime64(3)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list          = 'kafka1:9092,kafka2:9092',
    kafka_topic_list           = 'user-events',
    kafka_group_name           = 'ch-events-consumer',
    kafka_format               = 'JSONEachRow',
    kafka_num_consumers        = 4,
    kafka_skip_broken_messages = 100;

-- 실제 데이터를 저장할 영구 테이블
CREATE TABLE events_hourly_agg
(
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    hour         DateTime,
    event_count  AggregateFunction(count),
    unique_users AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event_type, country, hour)
TTL hour + INTERVAL 90 DAY;
 
-- Kafka Engine 테이블을 소스로 하는 Materialized View
CREATE MATERIALIZED VIEW mv_hourly TO events_hourly_agg AS
SELECT
    event_type,
    country,
    toStartOfHour(ts)  AS hour,
    countState()       AS event_count,
    uniqState(user_id) AS unique_users
FROM events_kafka
GROUP BY event_type, country, hour;

-- 쓰기: *State 함수로 중간 상태 저장 (Materialized View 내부)
SELECT
    countState()       AS event_count,
    uniqState(user_id) AS unique_users,
    sumState(revenue)  AS total_revenue
FROM events_kafka
GROUP BY event_type, country, hour;
 
-- 읽기: *Merge 함수로 상태를 최종 결과로 병합
SELECT
    event_type,
    country,
    hour,
    countMerge(event_count)   AS total_events,
    uniqMerge(unique_users)   AS dau,
    sumMerge(total_revenue)   AS revenue
FROM events_hourly_agg
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY event_type, country, hour
ORDER BY hour DESC;

-- 기본 테이블 ORDER BY: (event_type, country, hour)
-- country 기준 필터를 위한 Projection 추가
ALTER TABLE events_hourly_agg
    ADD PROJECTION proj_by_country
    (
        SELECT *
        ORDER BY (country, hour, event_type)
    );
 
-- 기존 데이터에 Projection 백필 적용
ALTER TABLE events_hourly_agg MATERIALIZE PROJECTION proj_by_country;
 
-- 이후 이 쿼리는 옵티마이저가 자동으로 proj_by_country를 선택
SELECT hour, countMerge(event_count) AS events
FROM events_hourly_agg
WHERE country = 'KR'
  AND hour >= now() - INTERVAL 7 DAY
GROUP BY hour;

SELECT query, projections
FROM system.query_log
WHERE query LIKE '%events_hourly_agg%'
  AND type = 'QueryFinish'
ORDER BY event_time DESC
LIMIT 5;

-- Kafka Engine 소스 테이블 (데이터를 저장하지 않는 스트리밍 소스)
CREATE TABLE events_kafka
(
    event_id    String,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    country     LowCardinality(String),
    ts          DateTime64(3)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list   = 'kafka1:9092,kafka2:9092',
    kafka_topic_list    = 'user-events',
    kafka_group_name    = 'ch-events-consumer',
    kafka_format        = 'JSONEachRow',
    kafka_num_consumers = 4;
    -- kafka_skip_broken_messages는 기본값(0)으로 두는 것을 권장
    -- 데이터 손실을 허용할 근거가 있을 때만 명시적으로 설정
 
-- 감사 로그용 raw 테이블
-- ReplicatedMergeTree는 ZooKeeper/ClickHouse Keeper가 구성된 클러스터 환경에서 사용
-- 단일 노드라면 MergeTree()로 대체
CREATE TABLE events_raw
(
    event_id    String,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    country     LowCardinality(String),
    ts          DateTime64(3)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events_raw', '{replica}')
ORDER BY (ts, event_type)
TTL ts + INTERVAL 30 DAY;
 
-- events_kafka → events_raw 연결 MV (원본 저장용)
CREATE MATERIALIZED VIEW mv_raw TO events_raw AS
SELECT *
FROM events_kafka;

CREATE TABLE events_hourly_agg
(
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    hour         DateTime,
    event_count  AggregateFunction(count),
    unique_users AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event_type, country, hour)
TTL hour + INTERVAL 90 DAY;
 
-- country 기준 필터 최적화를 위한 Projection
ALTER TABLE events_hourly_agg
    ADD PROJECTION proj_by_country
    (SELECT * ORDER BY (country, hour, event_type));
 
ALTER TABLE events_hourly_agg
    MATERIALIZE PROJECTION proj_by_country;

-- events_kafka → events_hourly_agg 연결 MV (집계용)
CREATE MATERIALIZED VIEW mv_hourly TO events_hourly_agg AS
SELECT
    event_type,
    country,
    toStartOfHour(ts)  AS hour,
    countState()       AS event_count,
    uniqState(user_id) AS unique_users
FROM events_kafka
GROUP BY event_type, country, hour;

-- 최근 24시간 이벤트 타입별 통계
SELECT
    event_type,
    countMerge(event_count)  AS total_events,
    uniqMerge(unique_users)  AS dau
FROM events_hourly_agg
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY event_type
ORDER BY total_events DESC;
 
-- 국가별 시간대 분석 (proj_by_country 자동 선택)
SELECT
    hour,
    countMerge(event_count) AS events
FROM events_hourly_agg
WHERE country = 'KR'
  AND hour >= now() - INTERVAL 7 DAY
GROUP BY hour
ORDER BY hour;

-- 일 단위 롤업 저장 테이블
CREATE TABLE events_daily_agg
(
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    day          Date,
    event_count  AggregateFunction(count),
    unique_users AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event_type, country, day);
 
-- 시간 단위 집계 결과를 소스로 하는 일 단위 롤업 뷰
CREATE MATERIALIZED VIEW mv_daily TO events_daily_agg AS
SELECT
    event_type,
    country,
    toDate(hour)                  AS day,
    countMergeState(event_count)  AS event_count,
    uniqMergeState(unique_users)  AS unique_users
FROM events_hourly_agg
GROUP BY event_type, country, day;

-- 전체 합산은 되지만 키별 분리가 없는 쿼리
-- countMerge() 자체는 올바른 값을 반환하지만, 이벤트 타입별 분리가 사라짐
SELECT countMerge(event_count) FROM events_hourly_agg;
 
-- 이벤트 타입별로 나눠보고 싶다면 GROUP BY 필수
SELECT event_type, countMerge(event_count)
FROM events_hourly_agg
GROUP BY event_type;
 
-- AggregateFunction 컬럼을 *Merge 없이 SELECT하면 바이너리 블롭 반환
SELECT event_count FROM events_hourly_agg;  -- 읽을 수 없는 값

-- 피해야 할 패턴 (숨겨진 내부 테이블 자동 생성)
CREATE MATERIALIZED VIEW mv_hourly AS SELECT ...;
 
-- 권장 패턴 (타겟 테이블 명시)
CREATE MATERIALIZED VIEW mv_hourly TO events_hourly_agg AS SELECT ...;