Count-Min Sketch

🔥 Vibe プロンプト

「10万アイテムのストリームで4KBメモリを使用して単語頻度をカウント。実際の値と比較。」

Count-Min Sketchとは？

Count-Min Sketch (CMS) は、ストリームデータ内の要素の出現頻度を推定するための確率的データ構造です。2次元のカウンター配列と複数のハッシュ関数を使用して、わずかなメモリで頻度を追跡します。

| 特性 | 値 | |------|-----| | メモリ | (e/ε) × ln(1/δ) 個のカウンター | | 誤差保証 | 確率 1-δ で、誤差 ≤ ε×N | | 方向性 | 常に過大評価（過小評価しない） |

CMSは常に過大評価します（ハッシュ衝突のため）。最小値を取ることで、最も正確な推定を得ます。

仕組み

1. 2次元テーブルを作成: 幅 = e/ε, 深さ = ln(1/δ)
2. 各行に独立したハッシュ関数を使用
3. アイテム追加: 各行でハッシュ → 対応カウンターをインクリメント
4. 頻度推定: 各行でハッシュ → 最小のカウンター値を取得

直感的理解:
- ハッシュ衝突によりCMSは過大評価する
- 各行は独立したハッシュ関数を使用
- 最小値を取ると、最も衝突の少ない行の値が選ばれる

数学的保証

確率 1-δ で:
  実際の頻度 ≤ 推定頻度 ≤ 実際の頻度 + ε × N

ここで:
  ε = エプシロン（精度パラメータ、例: 0.01）
  δ = デルタ（信頼度パラメータ、例: 0.01）
  N = 処理済みアイテム総数

例: ε=0.01, δ=0.01, N=100000の場合:
  幅 = e/0.01 ≈ 272
  深さ = ln(1/0.01) ≈ 5
  メモリ = 272×5 = 1,360 カウンター ≈ 5.3 KB
  誤差保証: ≤ 0.01×100000 = 1000

実装

import math
import mmh3

class CountMinSketch:
    def __init__(self, epsilon=0.01, delta=0.01):
        """
        epsilon: 誤差上限（全体の割合として）
        delta: 誤差が上限を超える確率
        """
        self.width = int(math.ceil(math.e / epsilon))
        self.depth = int(math.ceil(math.log(1 / delta)))
        self.table = [[0] * self.width for _ in range(self.depth)]
        self.total = 0
    
    def add(self, item, count=1):
        self.total += count
        for d in range(self.depth):
            idx = mmh3.hash(str(item), d) % self.width
            self.table[d][idx] += count
    
    def estimate(self, item):
        return min(
            self.table[d][mmh3.hash(str(item), d) % self.width]
            for d in range(self.depth)
        )
    
    def merge(self, other):
        """別のCMSをマージ（分散処理用）"""
        for d in range(self.depth):
            for w in range(self.width):
                self.table[d][w] += other.table[d][w]
        self.total += other.total

# テスト
import random
random.seed(42)

cms = CountMinSketch(epsilon=0.01, delta=0.01)
print(f"テーブルサイズ: {cms.width}x{cms.depth} = {cms.width*cms.depth} カウンター")

true_counts = {}
for _ in range(100000):
    item = f"item_{random.randint(1, 1000)}"
    cms.add(item)
    true_counts[item] = true_counts.get(item, 0) + 1

errors = []
for item, true_c in list(true_counts.items())[:10]:
    est = cms.estimate(item)
    errors.append(est - true_c)
    print(f"{item}: actual={true_c}, est={est}, error={est-true_c}")

print(f"\n平均誤差: {sum(errors)/len(errors):.2f}")
print(f"最大誤差: {max(errors)}")
print(f"理論保証: ≤ {0.01 * 100000} = 1000")

メモリ比較

| データ構造 | 100Kアイテム | メモリ | 誤差 | |-----------|-------------|-------|------| | HashMap | 100Kエントリ | ~5 MB | 0% | | CMS (1%誤差, 99%信頼) | 任意のストリーム | ~5 KB | ≤1% | | CMS (0.1%誤差, 99.9%信頼) | 任意のストリーム | ~50 KB | ≤0.1% |

実世界の応用

1. 人気商品ランキング

EコマースプラットフォームはCMSを使用してリアルタイムの商品人気度を追跡。100万アイテム/秒のストリームでも約5 KBでトレンド商品を特定できます。

2. DDoS検出

送信元IPごとのリクエスト数を監視。推定カウントが閾値を超えたIPをDDoS攻撃としてフラグします。

3. トレンドハッシュタグ

ソーシャルメディアプラットフォームはCMSを使用してハッシュタグの頻度をリアルタイム追跡。

4. ネットワークフロー監視

数百万の同時フローに対して、送信元IP、宛先IP、ポート組ごとの統計をCMSで管理。

5. IoTセンサー頻度分析

メモリ制限のあるエッジデバイスでCMSを使用してセンサー読み取り頻度をカウント。

時間複雑度

| 操作 | 時間 | |------|------| | add(item) | O(d) — d回のハッシュ + d個のインクリメント | | estimate(item) | O(d) — d回のハッシュ + d回の読み取り + 最小値 | | merge(other) | O(w×d) — 全カウンターの加算 | | 空間 | O(w×d) — εとδで設定可能 |

バリアント

| バリアント | 特徴 | |-----------|------| | Count-Min Sketch | 基本頻度推定 | | Count-Mean-Min Sketch | 他アイテムのノイズを差し引き | | Conservative Update | 最小値の場合のみインクリメント | | Heavy Hitters | CMS + ヒープでトップkアイテムを追跡 |

まとめ

頻度推定のための確率的データ構造
常に過大評価（最小値を取る）
設定可能: εが精度、δが信頼度を制御
少量メモリ: 1%誤差、99%信頼で約5 KB
マージ可能: 分散ストリーム処理に最適
使用例: Apache DataSketches, Redis, Twitter Algol
IoTセンサー頻度分析