5. Real-time Aggregation in Practice

1장의 요구로 돌아온다. 웹 성능 이벤트에서 최근 5분 동안의 LCP와 INP p75를 페이지별로 15초마다 계산하고, 기준을 넘으면 알림을 보내고 싶다. 앞단의 Kafka와 Flink는 그대로 두고 저장소만 바꾼다면, ClickHouse에서는 이것을 어떻게 만들고 BigQuery와는 무엇이 다를까.

5.1 Where the Latency Differs

이벤트가 알림이 되기까지의 지연을 네 구간으로 나눠 보면 비교할 곳이 분명해진다.

flowchart LR
  collect["① 수집<br/>브라우저에서 Kafka"] --> process["② 처리<br/>Flink"] --> load["③ 적재<br/>sink에서 DB"] --> query["④ 조회와 알림<br/>DB에서 alert"]

①과 ②는 저장소와 상관없이 같다. 비교할 구간은 ③과 ④다. 최종 알림 지연에는 평가 주기와 알림 정책도 더해진다.

처음에는 ③에서 ClickHouse가 유리하리라고 생각했다. 그런데 BigQuery도 Storage Write API의 committed stream을 쓰면 서버가 쓰기를 승인하는 즉시 데이터를 조회할 수 있다. 데이터가 보이기 시작하는 시점만 놓고 보면 둘 사이에 큰 차이가 있다고 하기 어렵다.

차이는 ④에 있다. 같은 작은 집계를 15초마다 반복하는 상황에서, ClickHouse에는 읽을 양을 미리 줄여 두는 장치가 있다. 2장에서 본 정렬 키와 희소 index, 그리고 증분 materialized view다.

이 표는 설계상 기대되는 차이다. 같은 워크로드로 두 시스템의 응답 시간을 실제로 재 보기 전까지는 가설일 뿐이다. 그래서 질문을 이렇게 바꿔 둔다. 같은 이벤트 스트림에서 5분 동안의 웹 성능 지표를 15초마다 집계한다면, ClickHouse가 BigQuery보다 더 낮고 안정적인 조회 지연을 낼까? 그렇다면 그 차이는 어떤 저장 설계와 집계 설계에서 나올까?

5.2 Design: Raw Table and 10-Second State Table

table은 두 개다. 원본 이벤트를 담는 table과, p75 상태를 10초 단위로 집계해 두는 table이다.

CREATE TABLE web_perf_raw (
  event_id   UUID,
  event_time DateTime64(3),
  metric     LowCardinality(String),
  page       LowCardinality(String),
  value      Float64
) ENGINE = MergeTree
PARTITION BY toDate(event_time)
ORDER BY (metric, page, event_time)
TTL toDateTime(event_time) + INTERVAL 7 DAY;

원본 table은 metric, page, event_time 순으로 정렬한다. metric과 page는 값의 종류가 적으므로 LowCardinality로 dictionary 인코딩한다.

정렬 키는 알림 쿼리가 거르는 순서에 맞춘다. 알림 쿼리는 metric을 같음 조건으로 고르고, 시간은 범위 조건으로 자른다. 같음 조건으로 거르는 컬럼을 앞에, 범위 조건으로 거르는 시간을 뒤에 두면 희소 index가 granule을 가장 많이 건너뛴다. 사이트가 여럿이라 늘 사이트로 먼저 거른다면 site를 맨 앞에 넣는다. 정렬 키는 table을 만들 때 정해지고 나중에 바꾸기 어렵다. 그래서 자주 쓰는 조회 패턴이 바뀌면 정렬 키부터 다시 검토해야 한다.

CREATE TABLE web_perf_10s (
  metric LowCardinality(String),
  bucket DateTime,
  page   LowCardinality(String),
  p75    AggregateFunction(quantileTDigest(0.75), Float64),
  n      AggregateFunction(count)
) ENGINE = AggregatingMergeTree
ORDER BY (metric, bucket, page)
TTL bucket + INTERVAL 30 DAY;

CREATE MATERIALIZED VIEW web_perf_10s_mv TO web_perf_10s AS
SELECT metric,
       toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket,
       page,
       quantileTDigestState(0.75)(value) AS p75,
       countState() AS n
FROM web_perf_raw
GROUP BY metric, bucket, page;

상태 table의 p75 컬럼은 값이 아니라 t-digest 상태를 담는다. 원본에 part가 들어올 때마다 materialized view가 그 행들만 10초 bucket으로 묶어 상태를 만들고, AggregatingMergeTree가 merge하면서 같은 bucket의 상태를 합친다. 상태 table은 시간 조건으로 조회하므로 bucket을 정렬 키의 두 번째에 둔다.

p75를 정확한 분위수 대신 t-digest로 근사하는 이유는 상태를 합칠 수 있어야 하기 때문이다. 정확한 분위수는 원래 값을 모두 들고 있어야 합칠 수 있지만, t-digest 요약은 작은 크기로 여러 개를 합칠 수 있다. 알림 판단에는 근사로 충분하다. 상태를 저장해 두므로 10초 bucket을 다시 묶어 1분, 1시간 단위의 p75도 계산할 수 있다.

보관 기간은 두 table을 다르게 둔다. 원본은 7일, 상태 table은 30일이다. 원본은 이벤트 한 건이 한 행이라 크고, 특정 이벤트를 다시 들여다보는 상세 분석에만 필요하다. 상태 table은 “metric과 페이지 조합 수 × 하루 8,640개 bucket"만큼만 커지므로 작고, 지난주와 이번 주의 p75를 비교하는 추세 확인에 오래 쓰인다. 원본이 TTL로 지워져도 이미 만든 상태는 남는다.

5.3 The Alert Query

알림 기준은 web.dev가 Core Web Vitals의 “good” 경계로 제시하는 값을 쓴다. 페이지 방문의 75번째 백분위수 기준으로 LCP는 2.5초, INP는 200ms다.

알림 쿼리는 상태를 합치기만 하고, 15초마다 실행된다.

SELECT metric,
       page,
       round(quantileTDigestMerge(0.75)(p75)) AS p75_ms,
       countMerge(n) AS n,
       transform(metric, ['LCP', 'INP'], [2500, 200], 0) AS threshold_ms
FROM web_perf_10s
WHERE metric IN ('LCP', 'INP')
  AND bucket >= now() - INTERVAL 5 MINUTE
GROUP BY metric, page
HAVING n >= 100 AND p75_ms > threshold_ms;

transform은 metric마다 다른 기준값을 붙인다. HAVING n >= 100은 표본이 적은 페이지를 거른다. 방문이 몇 건 안 되는 페이지는 p75가 크게 흔들리기 때문이다. bucket이 10초 단위라서 5분 창의 경계가 최대 10초 어긋날 수 있다. 15초 주기 알림이라면 받아들일 만한 오차다.

5.4 Alert Policy and Grafana

쿼리 결과에 한 번 잡혔다고 바로 알리지는 않는다. 같은 metric과 페이지가 두 번 연속 기준을 넘어야 알린다. 15초 주기에서 두 번 연속이면, 30초 안팎으로 유지된 악화만 알림이 된다. 순간적으로 튀는 값 하나에 알림이 울리는 것을 막는 장치다.

이 정책을 쿼리 안에 넣을 수도 있지만, 알림 도구에 맡기는 편이 단순하다. Grafana의 공식 ClickHouse 데이터 소스 플러그인은 Grafana-managed alerting을 지원한다. 알림 규칙의 쿼리로 SQL을 그대로 쓰면 된다.

Grafana에 넣을 때는 기준값 비교를 Grafana 쪽 threshold로 옮긴다. metric마다 기준이 다르므로, 쿼리가 p75를 기준값으로 나눈 비율을 돌려주게 하고 Grafana에서는 비율이 1을 넘는지만 본다.

SELECT metric,
       page,
       quantileTDigestMerge(0.75)(p75)
         / transform(metric, ['LCP', 'INP'], [2500, 200], 1) AS ratio
FROM web_perf_10s
WHERE metric IN ('LCP', 'INP')
  AND bucket >= now() - INTERVAL 5 MINUTE
GROUP BY metric, page
HAVING countMerge(n) >= 100;

결과의 문자열 컬럼인 metric과 page는 label이 되고, 숫자 컬럼인 ratio가 평가 대상이 된다. 그래서 행 하나가 알림 인스턴스 하나가 된다. 규칙의 설정은 이렇게 둔다.

5.5 Sink Settings

Flink sink가 batch를 충분히 크게 만들지 못하면, 2장에서 본 대로 part가 지나치게 많이 생긴다. 그래서 asynchronous insert를 켜서 행을 묶는 일을 서버에 맡긴다.

async_insert = 1
wait_for_async_insert = 1
async_insert_busy_timeout_ms = 1000

wait_for_async_insert = 1이 중요하다. 이 값을 끄면 서버는 데이터를 버퍼에 넣자마자 성공을 돌려준다. 켜 두면 버퍼의 데이터가 part로 쓰인 뒤에야 성공을 받는다. Flink의 checkpoint가 “이 offset까지는 적재됐다"고 믿으려면 성공이 실제 쓰기를 뜻해야 한다.

5.6 Measuring What Was Read

로컬 docker로 ClickHouse 24.8을 띄우고 위 table들을 만든 뒤, 합성 LCP 이벤트 520만 건을 넣었다. 그중 20만 건은 최근 5분에 몰아 넣고, 나머지는 하루에 걸쳐 흩어 놓았다. 그리고 같은 p75를 세 가지 방식으로 구한 뒤 system.query_log에서 각 쿼리가 읽은 양을 비교했다.

정렬 키만으로 읽는 행이 22분의 1로 줄었고, 상태 table을 쓰자 거기서 다시 170분의 1 가까이 줄었다. 상태 table에서 필요한 행은 원본 이벤트 수와 상관없이 “metric과 페이지 조합 수 × 5분 동안의 bucket 수(30개)“로 정해진다. 실제로는 granule 단위로 읽기 때문에 그보다 조금 더 읽지만, 이벤트가 늘어도 읽는 양은 거의 그대로다. 15초마다 반복해도 부담이 적은 이유다.

이 숫자는 응답 시간이 아니라 읽은 양이다. 로컬 한 대에서 합성 데이터로 잰 값이라 실제 지연과 비용은 알 수 없다. 그래도 2장에서 읽은 설계가 실제로 읽는 양을 이만큼 줄인다는 것은 확인할 수 있었다.

알림 정책까지 이어서 확인했다. 정상 이벤트를 보내다가 /item 페이지만 LCP와 INP가 나빠진 이벤트를 보내자, 첫 평가에서는 두 metric 모두 pending이 됐고 15초 뒤 다음 평가에서 firing으로 바뀌었다.

5.7 Late Events

브라우저 이벤트는 늦게 도착하기도 한다. 네트워크가 느리거나, 페이지를 떠날 때 모아 보내거나, Flink가 재시작하는 경우다.

이 설계는 bucket을 event_time으로 정한다. 그래서 2분 늦게 도착한 이벤트는 2분 전 bucket에 들어간다. materialized view는 새로 들어온 행으로 그 bucket의 상태를 하나 더 만들고, 조회나 merge 때 기존 상태와 합친다. t-digest 상태와 count 상태는 합치는 순서와 상관없이 같은 결과를 내므로, 지난 bucket에 나중에 더해도 문제가 없다. 직접 2분 전 시각으로 이벤트 200건을 넣어 보니 해당 bucket의 표본 수가 그만큼 늘었고, 다음 알림 평가에 바로 반영됐다. Flink에서 watermark로 창을 닫고 내보내는 방식과 달리, 늦은 이벤트를 버리지 않고 받아들일 수 있다.

남는 문제는 두 가지다.

• 5분보다 늦게 도착한 이벤트는 알림 창 밖의 bucket에 들어간다. 이미 지나간 평가는 다시 하지 않으므로 알림에는 영향을 주지 않는다. 대시보드나 추세 비교에는 그대로 반영된다.
• 가장 최근 bucket은 아직 이벤트가 다 도착하지 않은 상태다. 표본이 적은 상태로 p75를 계산하게 된다. 5분 창 전체로 보면 비중이 작고 n >= 100 조건도 있지만, 지연이 큰 환경이라면 창의 끝을 now() - INTERVAL 10 SECOND처럼 조금 당기는 방법도 있다.

5.8 What Is Still Missing: Duplicates in the View

이 설계에는 약점이 하나 더 있다. materialized view에 한 번 더해진 중복은 되돌릴 수 없다.

materialized view는 INSERT 시점에 계산된다. 그래서 원본 table을 나중에 ReplacingMergeTree로 바꿔 중복을 지우더라도, 그 전에 상태 table에 더해진 값은 그대로 남는다.

중복이 생기는 경로는 이렇다. Flink job이 실패해 마지막 checkpoint부터 다시 시작하면, checkpoint 이후에 이미 보냈던 이벤트를 다시 보낸다. 4장에서 본 hash 기반 중복 제거는 다시 보낸 묶음이 처음과 똑같을 때만 동작한다. 재시작 후에는 묶음 경계가 달라지기 쉽고, 그러면 다른 part로 보여 걸러지지 않는다.

기본값도 확인해 둘 필요가 있다. 24.8에서 확인한 값은 이렇다.

어떻게 대응할지는 지표의 성격에 따라 다르다. p75 알림은 약간의 중복에 크게 영향을 받지 않는다. 같은 이벤트가 몇 개 더 들어가도 분위수는 거의 움직이지 않으므로 그대로 둔다. 다만 중복은 표본 수 n을 부풀리므로, 표본 하한 조건이 실제보다 쉽게 통과될 수 있다는 점은 알아 둔다. 정확한 개수가 필요한 지표라면 Flink에서 event_id로 중복을 거르거나, checkpoint마다 insert_deduplication_token을 고정해서 같은 checkpoint의 재전송이 같은 묶음으로 인식되게 한다.

지난 sulco에서 Kafka와 Flink를 살펴보며 내린 결론이 여기서도 그대로 들어맞았다. 정확성은 sink에서 정해진다.

5.9 The Same Thing in BigQuery

비교를 위해 같은 모델을 BigQuery로 옮겨 보면 이렇다. 상태 table 없이 매번 원본을 집계한다. ClickHouse 쪽과 달리 이 SQL은 직접 실행해 보지 않았다.

CREATE TABLE my_dataset.web_perf_raw (
  event_id   STRING,
  event_time TIMESTAMP,
  metric     STRING,
  page       STRING,
  value      FLOAT64
)
PARTITION BY DATE(event_time)
CLUSTER BY metric, event_time
OPTIONS (partition_expiration_days = 7);

SELECT metric,
       page,
       APPROX_QUANTILES(value, 100)[OFFSET(75)] AS p75_ms,
       COUNT(*) AS n
FROM my_dataset.web_perf_raw
WHERE metric IN ('LCP', 'INP')
  AND event_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE)
GROUP BY metric, page
HAVING n >= 100
   AND p75_ms > CASE metric WHEN 'LCP' THEN 2500 WHEN 'INP' THEN 200 END;

날짜 partition으로 오늘 하루치만 남기고, clustering으로 그 안에서 metric과 시간 범위에 맞는 block만 읽기를 기대한다. clustering 컬럼은 앞에서부터 거를 때 효과가 크므로, 이번에는 page 대신 event_time을 두 번째에 뒀다.

이 쿼리는 쓰기에는 간단하지만, 15초마다 돌리면 확인할 것이 생긴다.

• 갓 들어온 데이터는 아직 clustering이 정리되지 않은 상태일 수 있다. 최근 5분만 읽는 쿼리에서 block이 실제로 얼마나 줄어드는지는 재 봐야 안다.
• on-demand 요금에서는 쿼리가 참조하는 table마다 최소 10MB를 청구한다. 15초 주기면 하루 5,760번 실행되므로, 읽는 양이 아주 작아도 이 최소량이 쌓인다.
• 사전 집계로 줄이려면 별도 설계가 필요하다. ClickHouse의 증분 materialized view처럼 INSERT 때 분위수 상태를 쌓아 두는 장치를 그대로 대응시키기는 어렵다.

5.10 Wrapping Up

ClickHouse는 나중에 읽을 양을 쓰는 순간에 미리 줄여 둔다. 정렬해서 쓰고, merge로 정리하고, materialized view로 미리 집계해 둔다. 그래서 같은 집계를 짧은 주기로 반복하는 쿼리에 강하다. 그 대가로 수정과 삭제, 중복 제거, 일관성은 나중으로 미뤄진다.

조회가 빠른 이유는 scan이 빨라서라기보다 읽지 않아도 되는 데이터를 건너뛰기 때문이다. 웹 성능 알림처럼 반복되는 작은 집계라면 이 설계가 잘 맞는다. 다만 중복, 늦게 도착한 이벤트, 집계의 정확성, 운영 부담을 어디서 감당할지는 따로 정해야 한다.

그래서 둘 중 하나를 고르기보다 역할을 나누는 구성을 생각해 볼 수 있다. BigQuery는 장기 보관과 폭넓은 분석을 맡고, ClickHouse는 자주 호출되는 실시간 집계와 알림을 맡는다. 같은 Flink 출력을 두 sink로 보내면 된다. 다만 두 저장소를 함께 운영할 가치가 있는지는 조회 지연, 비용, 유지보수 부담을 실제로 재 본 뒤에 정할 일이다.

6장에서는 이 설계를 로컬에서 직접 띄워 알림이 울리는 것까지 확인한다.