✍️ 필사 모드: 확률적 데이터 구조 완전 가이드 2025: HyperLogLog, Count-Min Sketch, MinHash, T-Digest, Cuckoo Filter — Redis/Presto/Spark 실전 분석

import mmh3  # MurmurHash3

class HyperLogLog:
    def __init__(self, p=14):
        self.p = p
        self.m = 1 << p  # 버킷 수
        self.buckets = [0] * self.m
        # 보정 상수 (p에 따라 다름)
        self.alpha_mm = self._get_alpha() * self.m * self.m

    def _get_alpha(self):
        if self.m == 16:
            return 0.673
        elif self.m == 32:
            return 0.697
        elif self.m == 64:
            return 0.709
        else:
            return 0.7213 / (1 + 1.079 / self.m)

    def add(self, item):
        x = mmh3.hash64(str(item))[0] & ((1 << 64) - 1)
        # 상위 p비트는 버킷 인덱스
        idx = x >> (64 - self.p)
        # 나머지 비트에서 선행 0 + 1 계산
        w = (x << self.p) & ((1 << 64) - 1)
        w |= 1 << (self.p - 1)  # 무한 루프 방지
        leading_zeros = 64 - w.bit_length() + 1
        self.buckets[idx] = max(self.buckets[idx], leading_zeros)

    def count(self):
        # 조화 평균 기반 추정
        z = sum(2.0 ** -b for b in self.buckets)
        estimate = self.alpha_mm / z

        # Small range correction (작은 값에서 편향 보정)
        if estimate <= 2.5 * self.m:
            zeros = self.buckets.count(0)
            if zeros > 0:
                estimate = self.m * (self.m / zeros).bit_length()  # 간이 log

        return int(estimate)

# 사용 예시
hll = HyperLogLog(p=14)
for i in range(1_000_000):
    hll.add(f"user_{i}")

print(f"Estimated: {hll.count()}")  # ~1,000,000 (오차 1% 이내)
print(f"Memory: {hll.m} buckets = {hll.m * 6 / 1024:.1f} KB")  # 12 KB

PFADD visitors:2025-04-15 user1 user2 user3
PFADD visitors:2025-04-15 user1  # 이미 있음, 카운트 증가 안 됨

PFCOUNT visitors:2025-04-15  # → 3

# 여러 날짜의 유니크 방문자 합계 (중복 제거)
PFCOUNT visitors:2025-04-15 visitors:2025-04-16 visitors:2025-04-17

# 병합
PFMERGE visitors:2025-04 visitors:2025-04-15 visitors:2025-04-16 ...
PFCOUNT visitors:2025-04  # → 월간 UV

import mmh3

class CountMinSketch:
    def __init__(self, width=2048, depth=5):
        self.width = width
        self.depth = depth
        self.table = [[0] * width for _ in range(depth)]
        self.seeds = [i * 31 + 7 for i in range(depth)]

    def add(self, item, count=1):
        for i in range(self.depth):
            idx = mmh3.hash(str(item), self.seeds[i]) % self.width
            self.table[i][idx] += count

    def estimate(self, item):
        return min(
            self.table[i][mmh3.hash(str(item), self.seeds[i]) % self.width]
            for i in range(self.depth)
        )

import heapq

class TopK:
    def __init__(self, k, cms):
        self.k = k
        self.cms = cms
        self.heap = []  # (count, item)
        self.items_in_heap = set()

    def add(self, item):
        self.cms.add(item)
        count = self.cms.estimate(item)

        if item in self.items_in_heap:
            # heap 재구성
            for i, (c, it) in enumerate(self.heap):
                if it == item:
                    self.heap[i] = (count, item)
                    heapq.heapify(self.heap)
                    return
        elif len(self.heap) < self.k:
            heapq.heappush(self.heap, (count, item))
            self.items_in_heap.add(item)
        elif count > self.heap[0][0]:
            removed = heapq.heappushpop(self.heap, (count, item))
            self.items_in_heap.remove(removed[1])
            self.items_in_heap.add(item)

    def top_k(self):
        return sorted(self.heap, reverse=True)

J(A, B) = |A ∩ B| / |A ∪ B|

import mmh3

class MinHash:
    def __init__(self, num_hashes=128):
        self.k = num_hashes
        self.seeds = list(range(num_hashes))
        self.signature = [float('inf')] * num_hashes

    def update(self, item):
        s = str(item)
        for i in range(self.k):
            h = mmh3.hash(s, self.seeds[i]) & 0xFFFFFFFF
            if h < self.signature[i]:
                self.signature[i] = h

    def jaccard(self, other):
        assert self.k == other.k
        matches = sum(1 for a, b in zip(self.signature, other.signature) if a == b)
        return matches / self.k

# 사용 예시
m1 = MinHash(num_hashes=128)
for word in ["apple", "banana", "cherry"]:
    m1.update(word)

m2 = MinHash(num_hashes=128)
for word in ["banana", "cherry", "date"]:
    m2.update(word)

print(m1.jaccard(m2))  # ≈ 0.5 (2/4)

from pyspark.ml.feature import MinHashLSH
from pyspark.ml.linalg import Vectors

data = spark.createDataFrame([
    (0, Vectors.sparse(6, [0, 1, 2], [1.0, 1.0, 1.0])),
    (1, Vectors.sparse(6, [2, 3, 4], [1.0, 1.0, 1.0])),
], ["id", "features"])

mh = MinHashLSH(inputCol="features", outputCol="hashes", numHashTables=5)
model = mh.fit(data)
model.approxSimilarityJoin(data, data, 0.6).show()

from sortedcontainers import SortedList

class TDigest:
    def __init__(self, compression=100):
        self.compression = compression
        self.centroids = SortedList(key=lambda c: c[0])  # (mean, count)
        self.total = 0

    def add(self, x, w=1):
        self.total += w
        if not self.centroids:
            self.centroids.add((x, w))
            return

        # 가장 가까운 centroid 찾기
        idx = self.centroids.bisect_left((x, 0))
        candidates = []
        if idx > 0:
            candidates.append(idx - 1)
        if idx < len(self.centroids):
            candidates.append(idx)

        closest = min(candidates, key=lambda i: abs(self.centroids[i][0] - x))
        c_mean, c_count = self.centroids[closest]

        # 이 centroid가 병합 가능한 최대 크기 (분위수 기반)
        q = self._quantile_of_centroid(closest)
        k = 4 * self.total * q * (1 - q) / self.compression

        if c_count + w <= k:
            new_count = c_count + w
            new_mean = (c_mean * c_count + x * w) / new_count
            del self.centroids[closest]
            self.centroids.add((new_mean, new_count))
        else:
            self.centroids.add((x, w))

    def quantile(self, q):
        """q번째 분위수 반환."""
        if not self.centroids:
            return None

        target = q * self.total
        cum = 0
        for mean, count in self.centroids:
            if cum + count / 2 >= target:
                return mean
            cum += count
        return self.centroids[-1][0]

    def _quantile_of_centroid(self, idx):
        cum = 0
        for i, (_, c) in enumerate(self.centroids):
            if i == idx:
                return (cum + c / 2) / self.total
            cum += c
        return 1.0

import mmh3

class CuckooFilter:
    def __init__(self, capacity, bucket_size=4, max_kicks=500):
        self.buckets = [[] for _ in range(capacity)]
        self.bucket_size = bucket_size
        self.max_kicks = max_kicks
        self.capacity = capacity

    def _fingerprint(self, item):
        return mmh3.hash(str(item)) & 0xFF  # 8비트 fingerprint

    def _index(self, fp, bucket):
        return (bucket ^ mmh3.hash(str(fp))) % self.capacity

    def add(self, item):
        fp = self._fingerprint(item)
        i1 = mmh3.hash(str(item)) % self.capacity
        i2 = self._index(fp, i1)

        if len(self.buckets[i1]) < self.bucket_size:
            self.buckets[i1].append(fp)
            return True
        if len(self.buckets[i2]) < self.bucket_size:
            self.buckets[i2].append(fp)
            return True

        # Eviction loop
        i = i1  # 또는 i2를 골라도 됨
        for _ in range(self.max_kicks):
            # 랜덤 fingerprint 쫓아내기
            import random
            idx = random.randint(0, self.bucket_size - 1)
            fp, self.buckets[i][idx] = self.buckets[i][idx], fp
            i = self._index(fp, i)
            if len(self.buckets[i]) < self.bucket_size:
                self.buckets[i].append(fp)
                return True

        return False  # 가득 참

    def contains(self, item):
        fp = self._fingerprint(item)
        i1 = mmh3.hash(str(item)) % self.capacity
        i2 = self._index(fp, i1)
        return fp in self.buckets[i1] or fp in self.buckets[i2]

    def remove(self, item):
        fp = self._fingerprint(item)
        i1 = mmh3.hash(str(item)) % self.capacity
        i2 = self._index(fp, i1)

        if fp in self.buckets[i1]:
            self.buckets[i1].remove(fp)
            return True
        if fp in self.buckets[i2]:
            self.buckets[i2].remove(fp)
            return True
        return False

# HyperLogLog
PFADD mykey a b c
PFCOUNT mykey

# Bloom Filter (RedisBloom 모듈)
BF.RESERVE myfilter 0.01 1000000  # 1% 오차, 100만 개
BF.ADD myfilter user1
BF.EXISTS myfilter user1

# Count-Min Sketch
CMS.INITBYPROB mycms 0.001 0.01
CMS.INCRBY mycms item 1
CMS.QUERY mycms item

# T-Digest
TDIGEST.CREATE mydigest
TDIGEST.ADD mydigest 1.5 2.5 3.5 ...
TDIGEST.QUANTILE mydigest 0.99

# Top-K
TOPK.RESERVE mytopk 10 1000 3 0.9
TOPK.ADD mytopk item1 item2 ...
TOPK.LIST mytopk

# Spark SQL
df.agg(F.approx_count_distinct("user_id").alias("unique_users"))  # HyperLogLog
df.agg(F.percentile_approx("latency", 0.99))  # T-Digest

# PySpark MLlib
from pyspark.ml.feature import MinHashLSH

SELECT approx_distinct(user_id) FROM events;  -- HyperLogLog
SELECT approx_percentile(latency, 0.99) FROM requests;  -- Quantile Digest

-- HyperLogLog 병합
SELECT cardinality(merge(cast(hll_sketch AS HyperLogLog)))
FROM daily_hlls;

SELECT uniqExact(user_id);       -- 정확한 distinct
SELECT uniqHLL12(user_id);       -- HyperLogLog
SELECT quantileTDigest(0.99)(latency);