Count-Min Sketch：Twitter のトレンドを支える頻度推定アルゴリズム

import hashlib
class CountMinSketch:    def __init__(self, width=2048, depth=5):        self.w, self.d = width, depth        self.table = [[0] * width for _ in range(depth)]        # 異なるハッシュ関数のための seed        self.hashes = [hashlib.sha256(str(i).encode()).digest() for i in range(depth)]
    def _idx(self, item, i):        h = hashlib.blake2b(str(item).encode() + self.hashes[i], digest_size=4).digest()        return int.from_bytes(h, 'big') % self.w
    def add(self, item, count=1):        for i in range(self.d):            self.table[i][self._idx(item, i)] += count
    def count(self, item):        # 全ハッシュ層の最小値 (上振れを抑えるため)        return min(self.table[i][self._idx(item, i)] for i in range(self.d))
    def merge(self, other):        '''複数の CMS を結合 (分散集計)'''        for i in range(self.d):            for j in range(self.w):                self.table[i][j] += other.table[i][j]
# テスト: 8 単語のストリームを 100,000 件処理import randomrandom.seed(0)words = ['the', 'quick', 'brown', 'fox', 'jumps', 'over', 'lazy', 'dog']weights = [100, 50, 30, 30, 20, 20, 10, 5]
cms = CountMinSketch(width=512, depth=5)true = {}for _ in range(100_000):    w = random.choices(words, weights=weights)[0]    cms.add(w)    true[w] = true.get(w, 0) + 1
print('Word     | True   | CMS    | Error')for w in words:    est = cms.count(w)    err = abs(est - true[w]) / true[w] * 100    print(f'  {w:8s} | {true[w]:6d} | {est:6d} | {err:.2f}%')
# 実機実行結果: メモリ ~10KB、誤差 0% (この規模)

import heapq
class TopK:    def __init__(self, k=10, cms_width=2048, cms_depth=5):        self.k = k        self.cms = CountMinSketch(width=cms_width, depth=cms_depth)        self.heap = []  # (count, item) の最小ヒープ        self.in_heap = set()
    def add(self, item):        self.cms.add(item)        c = self.cms.count(item)        if item in self.in_heap:            # 既に heap にあれば再構築            for i, (cnt, it) in enumerate(self.heap):                if it == item:                    self.heap[i] = (c, it)                    heapq.heapify(self.heap)                    return        if len(self.heap) < self.k:            heapq.heappush(self.heap, (c, item))            self.in_heap.add(item)        elif c > self.heap[0][0]:            old = heapq.heapreplace(self.heap, (c, item))            self.in_heap.discard(old[1])            self.in_heap.add(item)
    def top(self):        return sorted(self.heap, key=lambda x: -x[0])