PySpark 고카디널리티 집계 — count(distinct) 가 클러스터를 멈출 때

투표함을 상상해 보세요. "투표자가 몇 명인가?"를 알려면 중복 투표를 걸러내야 하므로, 투표한 사람 전체 목록을 메모리에 들고 비교해야 합니다. SELECT count(distinct user_id) 가 하는 일이 정확히 이것입니다. 그래서 수억 사용자의 UV 집계는 고카디널리티 distinct 집계 — 대규모 데이터에서 가장 비싼 연산 중 하나가 됩니다.

이 글은 왜 distinct 집계가 비싼지, 그리고 근사 집계(HyperLogLog), 2단계 집계, 사전 집계 롤업으로 이 문제를 푸는 법을 정리합니다.

이 글에서 배우는 것

• count(distinct) 가 왜 count(*) 와 비교도 안 되게 비싼지
• HyperLogLog(approx_count_distinct)로 메모리를 고정 크기로 유지하는 원리
• 스큐한 그룹의 집계를 2단계 salt 패턴으로 분산하는 방법
• 사전 집계 롤업에서 distinct 를 단순히 더하면 안 되는 이유(가산성 함정)
• collect_set OOM 과 언제 어떤 해법을 써야 하는지

1. 왜 count(distinct) 는 비싼가

count(*)              → 단순 카운터 1개 (싸다)
count(distinct x)     → 본 모든 고유값을 추적해야 함 (비싸다)

정확한 distinct 는 "이미 본 값인가?"를 판단하기 위해 고유값 집합을 메모리에 유지합니다. 카디널리티가 수억이면 그 집합 자체가 거대해집니다. 게다가 여러 컬럼으로 group 하면, 그룹마다 고유값 집합이 필요해 메모리가 곱으로 늘어납니다. 비용이 O(고유값 수 × 그룹 수) 로 폭발하는 이유입니다.

# 그룹 × 고카디널리티 → 메모리 폭발
df.groupBy("country", "date").agg(F.countDistinct("user_id"))
# country×date 그룹마다 user_id 고유값 집합 유지

2. 해법 ① approx_count_distinct (HyperLogLog)

대부분의 분석 지표(UV, 도달 수)는 약간의 오차가 허용됩니다. 1,000,234 명이든 1,001,050 명이든 비즈니스 판단은 같습니다. 여기서 "정확함"을 고집하면 엄청난 비용을 치릅니다. 이럴 때 HyperLogLog 기반 근사 집계가 정답입니다.

from pyspark.sql import functions as F
 
# 정확 (비쌈)
df.groupBy("country").agg(F.countDistinct("user_id").alias("uv"))
 
# 근사 (싸고 빠름, 기본 오차 ~5%)
df.groupBy("country").agg(F.approx_count_distinct("user_id").alias("uv"))
 
# 오차율 지정 (rsd: relative standard deviation, 작을수록 정확·메모리↑)
df.groupBy("country").agg(F.approx_count_distinct("user_id", rsd=0.02).alias("uv"))

HyperLogLog 는 고유값 집합 전체가 아니라 고정 크기의 작은 스케치(sketch) 만 유지합니다. 카디널리티가 수억이어도 메모리가 일정합니다.

판단 기준: 정확한 값이 꼭 필요한가? 과금·정산이면 정확히, 대시보드 지표·추세면 근사로. 근사 하나로 잡이 수십 배 빨라지는 경우가 흔합니다.

3. 해법 ② 2단계 집계 (Partial → Final)

Spark 는 기본적으로 집계를 2단계(map-side partial → reduce-side final)로 합니다. 그런데 특정 그룹에 데이터가 심하게 몰려 있다면(스큐), final 단계에서 한 리듀서가 혼자 모든 짐을 집니다. 이를 막기 위해 salt 로 인위적으로 단계를 나눕니다.

# 스큐한 그룹 키에 salt 를 붙여 부분 집계 → salt 제거 후 최종 집계
N = 16
salted = df.withColumn("salt", (F.rand() * N).cast("int"))
 
partial = (salted
    .groupBy("country", "salt")          # salt 로 분산해 부분 집계
    .agg(F.sum("amount").alias("partial_sum")))
 
final = (partial
    .groupBy("country")                  # salt 제거하고 합산
    .agg(F.sum("partial_sum").alias("total")))

sum, count, max 같은 결합법칙이 성립하는(associative) 집계는 이렇게 2단계로 안전하게 나눌 수 있습니다. (스큐 일반론은 별도 글 "PySpark 데이터 스큐 완전 정복".)

주의: count(distinct) 는 단순 2단계 합산이 안 됩니다(부분 distinct 를 더하면 틀림). distinct 스큐는 approx 또는 아래 사전 집계로 접근하세요.

4. 해법 ③ 사전 집계 롤업 (Pre-aggregation)

대시보드가 매 조회마다 수십억 행 원본을 스캔한다면 비효율적이죠. 같은 집계를 반복 조회한다면 원본을 매번 스캔하지 말고 미리 집계해 작은 롤업 테이블로 만들어 두는 것이 훨씬 효율적입니다.

# 일별·국가별 지표를 미리 집계해 저장 (배치)
daily = (events
    .groupBy("date", "country")
    .agg(
        F.count("*").alias("events"),
        F.approx_count_distinct("user_id").alias("uv"),
        F.sum("amount").alias("revenue")))
 
daily.writeTo("analytics.daily_metrics").append()
 
# 대시보드는 작은 롤업만 조회 (원본 스캔 없음)

월별·연별 지표는 일별 롤업을 다시 집계하면 됩니다(원본 재스캔 불필요).

가산성(Additivity) 함정

롤업을 재집계할 때 조심해야 할 함정이 있습니다. distinct 는 단순히 더할 수 없습니다. 월요일 UV 100명, 화요일 UV 80명을 더한다고 이틀간 UV 가 180명이 아닙니다(중복 사용자가 있으니까요). 해결책은 HLL 스케치를 저장해 나중에 병합하는 것입니다.

# Spark 의 sketch 함수로 병합 가능한 스케치를 저장 (구현/버전에 따라 datasketches 활용)
# 일별 스케치 저장 → 월별엔 스케치들을 union 하여 distinct 추정

5. 해법 ④ collect_set 폭발 피하기

"그룹별 고유값 목록"이 필요해 collect_set 을 쓰다 보면, 그룹의 고유값이 많을 때 한 행에 거대한 배열이 생겨 OOM 이 납니다. 정말 목록이 필요한지 먼저 확인하세요.

# 위험: 그룹당 수백만 고유값을 한 배열에
df.groupBy("country").agg(F.collect_set("user_id"))
 
# 개수만 필요하면 collect_set 대신 distinct count
df.groupBy("country").agg(F.approx_count_distinct("user_id"))
 
# 목록이 정말 필요하면 크기 제한 또는 별도 분해 저장

collect_list/collect_set 은 결과가 한 행·한 익스큐터에 모이므로, 그룹 카디널리티가 크면 위험합니다.

6. 진단 — 어디서 터지나

Spark UI 에서 final 집계 스테이지의 태스크 스큐·스필을 확인하세요(별도 글 "PySpark 느린 잡 디버깅").

7. 정리

고카디널리티 집계의 핵심 통찰은 "정확함이 정말 필요한지 먼저 묻는 것"입니다. distinct 의 정확한 값은 비싸지만, 대부분의 분석 지표는 HyperLogLog 근사로 충분합니다. 여기에 가산적 집계의 2단계 처리와 사전 집계 롤업을 더하면, 수억 카디널리티의 집계도 일정한 메모리로 안정적으로 처리할 수 있습니다. "count distinct 가 클러스터를 멈춘다"는 더 이상 어쩔 수 없는 일이 아닙니다.

마치며 — 핵심 요약

• count(distinct) 는 그룹마다 고유값 집합을 메모리에 유지해야 합니다. 그룹 수 × 카디널리티만큼 비쌉니다.
• 분석 지표(UV·도달)에는 approx_count_distinct(HyperLogLog)가 정답입니다. 메모리가 고정 크기로 유지됩니다.
• 가산적 집계(sum·count)의 스큐는 salt 2단계 로 분산할 수 있지만, distinct 는 안 됩니다.
• 롤업 테이블에서 distinct 를 재집계하려면 HLL 스케치를 저장하고 병합해야 합니다. 단순 합산은 틀립니다.
• collect_set/collect_list 는 그룹 카디널리티가 크면 OOM 의 씨앗입니다. 개수만 필요하면 approx_count_distinct 로 대체하세요.
• 한 번만 생각해 두면 — "정확해야 하는가, 근사여도 되는가?" — 집계 설계의 절반이 해결됩니다.

이 글은 Spark 3.5 기준으로 작성되었습니다. 대규모 지표 집계·롤업 파이프라인 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀