대시보드의 통계 API는 처음에 원본 이벤트와 결과 항목 테이블을 요청마다 집계했습니다. 정확하고 구현이 단순했지만 데이터가 늘면서 하나의 화면이 여러 개의 GROUP BY를 동시에 실행했습니다. 기간, 장비, 제품, 결함 종류 조합이 다양해 캐시 적중률도 낮았습니다.
짧은 기간은 괜찮았지만 장기 조회와 동시 사용자가 겹치면 응답 시간이 크게 흔들렸습니다. 쿼리 튜닝과 인덱스만으로는 한계가 있었습니다. 사용자가 요구하는 결과는 매번 원본 수백만 건을 다시 읽어야 할 만큼 자유롭지 않았습니다.
대부분 시간 버킷별 count와 비율을 다시 합치는 형태였습니다. 그래서 원본 적재 시 통계 delta를 계산해 bucket table에 미리 누적하는 구조로 바꿨습니다. 사전 계산은 단순히 속도를 위해 정확성을 포기하는 방식이 아닙니다.
설계할 항목은 네 가지였습니다. 중복 반영 방지, 버킷 경계, 지연 데이터, 원본 대조였습니다.
통계 차원을 먼저 줄인 이유
원본 데이터에는 필드가 많지만 모든 조합을 사전 계산하면 차원 조합 수가 폭발적으로 늘어납니다. 그래서 대시보드와 API가 실제로 사용하는 차원부터 목록으로 만들었습니다.
- 시간 단위와 bucket 시작 시각.
- source, device, category.
- 오류 발생 여부.
- finding 이름과 종류.
- segment의 PC, camera, index.
- confidence level과 모델 배포 코드.
모든 통계를 한 테이블에 넣지 않았습니다. event count, finding count, segment count, confidence 분포처럼 고유 키와 값의 의미가 다른 집계를 분리했습니다.
data class EventBucketDelta(
val unit: StatisticsUnit,
val bucketStart: Instant,
val bucketEnd: Instant,
val sourceKey: String,
val deviceKey: String,
val categoryKey: String,
val hasIssue: Boolean,
val count: Long,
val boundaryCount: Long,
)
미리 계산할 차원을 선택할 때 쓰기 증폭과 조회 비용을 함께 봤습니다. 거의 사용하지 않는 고차원 조합은 원본 또는 더 거친 bucket에서 계산하고, 반복 조회되는 핵심 조합만 저장했습니다.
공통 도메인 규칙이 된 시간 버킷
버킷 경계가 Worker와 조회 API에서 다르면 숫자가 맞지 않습니다. floor(epoch / unit) * unit과 같은 규칙을 공통 타입에 두고 모든 시각을 UTC Instant로 계산했습니다.
enum class StatisticsUnit(val minutes: Long) {
MINUTE_10(10),
HOUR(60),
DAY(1_440),
}
fun StatisticsUnit.window(time: Instant): Pair<Instant, Instant> {
val size = Duration.ofMinutes(minutes).seconds
val start = Instant.ofEpochSecond(time.epochSecond / size * size)
return start to start.plusSeconds(size)
}
일간 통계를 사용자 로컬 자정으로 보여줘야 한다면 UTC 24시간 버킷과 달력 일자를 구분해야 합니다. DST가 있는 지역에서 하루는 항상 24시간이 아닙니다. 저장 버킷은 UTC 기반으로 고정하고, 달력 기간 비교가 필요하면 조회 planner가 지역 시간 경계를 UTC window로 변환하도록 분리했습니다.
경계 테스트에는 버킷 시작 바로 전·정확히 시작·끝 직전·정확히 끝을 포함했습니다. 반개구간 [start, end)을 사용해 한 이벤트가 정확히 하나의 버킷에만 들어가게 했습니다.
같은 이벤트를 두 번 누적하지 않게 했다
Redis Streams 전달과 ACK 경계는 at-least-once입니다. Worker가 통계 upsert 후 ACK 전에 종료되면 같은 이벤트가 다시 처리됩니다. count = count + 1을 단순 재실행하면 통계가 원본보다 커집니다.
통계 처리 전에 이벤트 ID를 processed table에 조건부 insert했습니다. 처음 보는 ID일 때만 bucket delta를 만듭니다. 이 insert와 bucket upsert는 같은 PostgreSQL transaction 안에서 실행합니다.
INSERT INTO processed_statistics_event (event_id, processed_at) VALUES (:event_id, now()) ON CONFLICT (event_id) DO NOTHING RETURNING event_id;
val unprocessed = items.mapNotNull { item ->
if (processedRepository.markIfAbsent(item.eventId)) {
StatisticsPrecompute.of(item)
} else {
duplicateCounter.increment()
null
}
}
processed marker만 커밋되고 bucket 갱신이 롤백되는 부분 성공을 막기 위해 MANDATORY transaction 안에서 실행했습니다. 호출하는 이벤트 처리 서비스가 시작한 transaction을 반드시 공유하도록 한 것입니다.
Batch 안에서 먼저 합치는 동일 키 delta
한 batch에 같은 source, device, category, bucket으로 들어가는 이벤트가 여러 개 있을 수 있습니다. 각 이벤트마다 개별 upsert를 보내면 같은 PostgreSQL row를 반복해서 잠그고 round trip이 늘어납니다. 메모리에서 count 필드를 제외한 키로 먼저 group한 뒤 합산했습니다.
fun List<EventBucketDelta>.aggregate(): List<EventBucketDelta> =
groupBy { it.copy(count = 0, boundaryCount = 0) }
.map { (key, values) ->
key.copy(
count = values.sumOf { it.count },
boundaryCount = values.sumOf { it.boundaryCount },
)
}
정렬 순서도 고정했습니다. 여러 transaction이 같은 bucket row들을 서로 다른 순서로 update하면 deadlock 가능성이 커집니다. unit, bucket start, source, device, category와 세부 차원 순으로 정렬한 뒤 batch upsert했습니다.
Upsert는 delta를 더하는 방식으로 만들었다
각 bucket row의 고유 키는 unit, bucket start, 업무 차원 조합입니다. 처음이면 insert하고 이미 있으면 기존 count에 이번 batch delta를 더합니다.
INSERT INTO event_statistics_bucket (
bucket_start, bucket_end, bucket_minutes,
source_key, device_key, category_key, has_issue,
count, boundary_count
) VALUES (
:bucket_start, :bucket_end, :bucket_minutes,
:source_key, :device_key, :category_key, :hasIssue,
:count, :boundary_count
)
ON CONFLICT (
bucket_minutes, bucket_start,
source_key, device_key, category_key, has_issue
)
DO UPDATE SET
count = event_statistics_bucket.count + EXCLUDED.count,
boundary_count = event_statistics_bucket.boundary_count + EXCLUDED.boundary_count,
updated_at = now();
한 SQL에 지나치게 많은 parameter를 만들지 않도록 delta 목록도 chunk로 나눴습니다. 적절한 chunk 크기는 네트워크 왕복, statement 크기, lock 보유 시간, JDBC 메모리를 함께 측정해 정했습니다.
| 방식 | 읽기 비용 | 쓰기 비용 | 정확성 관리 |
|---|---|---|---|
| 요청마다 원본 집계 | 데이터량에 비례 | 추가 없음 | 원본과 항상 동일 |
| 결과 캐시 | cache hit이면 낮음 | 무효화 필요 | TTL 동안 오래된 값 가능 |
| 사전 계산 bucket | bucket 수에 비례 | 매 이벤트 delta upsert | 중복·재계산 경로 필요 |
| Materialized View | refresh 후 낮음 | refresh 비용 큼 | refresh 시점 기준 |
실시간에 가까운 대시보드와 at-least-once pipeline에는 incremental bucket이 적합했습니다. 완전한 실시간성이 필요 없고 refresh window를 허용한다면 materialized view가 더 단순할 수 있습니다.
조회는 완성된 버킷을 합치고 경계를 조심했다
10분 버킷으로 1시간 통계를 조회하면 해당 범위의 count를 합하면 됩니다. 하지만 요청 시작과 끝이 버킷 중간에 걸리면 전체 버킷을 더할 경우 범위 밖 데이터가 포함됩니다. 두 가지 정책 중 하나를 명확히 해야 합니다.
- API가 버킷 경계로 정렬된 기간만 허용합니다.
- 완전히 포함된 내부 버킷은 사전 계산에서 읽고 양쪽 부분 구간은 원본에서 계산합니다.
두 번째 방식은 정확하지만 조회가 복잡합니다. 대시보드가 정해진 캘린더 기간을 사용하므로 요청 window를 정책 객체가 버킷 경계에 맞게 계획하도록 했습니다. 사용자가 임의 시각을 입력하는 API는 실제 제공 정확도를 문서화해야 합니다.
SELECT bucket_start,
SUM(count) AS event_count,
SUM(boundary_count) AS boundary_count
FROM event_statistics_bucket
WHERE bucket_start >= :start_at
AND bucket_start < :end_at
AND bucket_minutes = :unit_minutes
AND source_key = :source_key
GROUP BY bucket_start
ORDER BY bucket_start;
늦게 도착한 데이터와 수정 데이터의 구분
이벤트 발생 시각이 과거인 데이터가 늦게 도착해도 해당 과거 bucket에 delta를 더하면 됩니다. Parquet Cold 계층으로 이미 이관된 범위라면 문제가 달라집니다. Hot bucket만 바꾸면 Cold 파일과 숫자가 다르게 됩니다.
허용 late arrival 기간을 Hot 보존 범위 안에 두거나, Cold correction manifest를 만드는 정책이 필요합니다. 변경 가능한 원본을 지원한다면 단순 delta +1뿐 아니라 이전 값을 빼고 새 값을 더하는 보정 이벤트 또는 bucket 재계산이 필요합니다. 현재 구조는 수집 후 원본이 사실상 불변이라는 전제에 적합합니다.
이 전제가 바뀌면 processed marker만으로는 충분하지 않고 이벤트 버전과 보정 전략을 추가해야 합니다.
원본과의 정기 대조
사전 계산은 한 번 잘못 누적되면 이후 조회가 빠르게 잘못된 숫자를 반환합니다. 테스트뿐 아니라 운영에서 샘플 기간을 원본 집계와 비교하는 검증이 필요합니다.
- 원본 이벤트 수와 bucket 합계 차이.
- processed marker 수와 원본 고유 이벤트 수.
- 중복 skip counter 증가율.
- table·unit별 delta 개수와 upsert 시간.
- batch 전체 처리 시간과 slow batch 로그.
- bucket partition 생성 실패.
- 과거 bucket의 예상치 못한 갱신.
차이가 발견되면 범위별 bucket을 삭제하고 원본에서 다시 계산하는 backfill 도구가 있어야 합니다. 사전 계산 테이블을 복구 불가능한 진실로 만들지 않는 것이 중요합니다.
중복 없이 증분 집계를 유지하는 조건
실시간 집계가 느려졌을 때 가장 먼저 한 일은 캐시를 더 붙이는 것이 아니라 사용자가 반복해서 보는 집계 단위를 찾는 일이었습니다. 시간과 업무 차원을 bucket key로 만들고 원본 적재 transaction 안에서 delta를 누적하자 조회 비용이 원본 행 수가 아니라 bucket 수에 비례하게 됐습니다. 대신 중복 처리 방지, 버킷 경계, 늦게 도착한 데이터, 재계산이라는 책임이 생겼습니다.
processed marker와 bucket upsert를 같은 transaction에 넣고, batch 안에서 delta를 합치며, 원본 대조 경로를 유지해 이 비용을 통제했습니다. 사전 계산은 성능 최적화인 동시에 새로운 데이터 모델입니다. 최신성, 수정 가능성, 시간 경계와 복구 방법을 명시할 수 있을 때만 빠른 숫자가 신뢰할 수 있는 숫자가 됩니다.
답글 남기기