確率的データ構造入門：Bloom FilterとHyperLogLogのPython実装

確率的データ構造とは

大規模データを扱う場面では、すべての要素を正確に記録しようとするとメモリが膨大になります。確率的データ構造（Probabilistic Data Structure）は、わずかな誤差を許容する代わりに、メモリ使用量を大幅に削減するデータ構造の総称です。

確率的データ構造の基本的なトレードオフは次のとおりです。

本記事では、代表的な確率的データ構造である Bloom Filter（集合への所属判定）と HyperLogLog（ユニーク要素数の推定）を取り上げ、理論的な背景とともに Python によるスクラッチ実装を紹介します。

Bloom Filter

Bloom Filter は、ある要素が集合に含まれるかどうかを高速に判定するための確率的データ構造です。1970 年に Burton Howard Bloom が提案しました。

仕組み

Bloom Filter は以下の 2 つの要素で構成されます。

• ビット配列: サイズ \(m\) の配列（すべて 0 で初期化）
• ハッシュ関数: \(k\) 個の独立なハッシュ関数 \(h_1, h_2, \dots, h_k\) （各関数は \(\{0, 1, \dots, m-1\}\) の値を返す）

挿入操作（add）: 要素 \(x\) を追加するとき、\(k\) 個のハッシュ関数を適用し、対応するビットをすべて 1 にします。

検索操作（contains）: 要素 \(x\) が含まれるかを判定するとき、\(k\) 個のハッシュ値に対応するビットを確認します。

• すべてのビットが 1 → 「おそらく含まれる」（偽陽性の可能性あり）
• 1 つでもビットが 0 → 「確実に含まれない」（偽陰性は発生しない）

この性質が Bloom Filter の最大の特徴です。偽陰性（False Negative）は絶対に発生しませんが、偽陽性（False Positive）は発生し得ます。

偽陽性率の理論

\(n\) 個の要素を挿入した後の偽陽性率は、次の式で近似されます。

ここで \(m\) はビット配列のサイズ、\(k\) はハッシュ関数の個数、\(n\) は挿入済み要素数です。

偽陽性率を最小化する最適なハッシュ関数の個数は次のとおりです。

このとき偽陽性率は次のように簡略化されます。

これらの式を天下り的に受け入れるのではなく、以下で導出過程を確認します。

導出 1：偽陽性率の式（式 1）

\(k\) 個のハッシュ関数が各要素に対して独立かつ一様にビット位置を選ぶと仮定します（この独立性の仮定は近似であり、後述のエッジケースで限界を議論します）。

ステップ 1: 1 回のハッシュ適用である特定のビットが 0 のままである確率

ビット配列のサイズが \(m\) のとき、1 回のハッシュ適用が特定の 1 ビットを選ばない確率は \(1 - 1/m\) です。

ステップ 2: \(n\) 個の要素を挿入した後、特定のビットが 0 のままである確率

\(n\) 個の要素それぞれに \(k\) 個のハッシュ関数を適用するので、合計 \(kn\) 回のハッシュ適用が行われます。したがって

ここで、極限 \(\left(1 - \frac{1}{m}\right)^m \to e^{-1}\) （\(m \to \infty\) ）を用いると

ステップ 3: 特定のビットが 1 である確率

ステップ 4: 偽陽性率

未挿入の要素を検索するとき、\(k\) 個のハッシュ位置すべてがたまたま 1 になっている確率が偽陽性率です。各ビットが独立に 1 になっていると近似すると

となり、式 (1) が得られます。

導出 2：最適なハッシュ関数の個数（式 2）

式 (1) を \(k\) について最小化します。\(u = e^{-kn/m}\) とおくと、\(P(fp) = (1-u)^k\) かつ \(kn/m = -\ln u\) です。対数を取ると

\(k\) について微分し（\(u\) も \(k\) に依存することに注意し、連鎖律を適用）、\(0\) とおくと

\(k n/m = -\ln u\) を代入し、両辺に \((1-u)\) を掛けて整理すると

この式は \(u \leftrightarrow 1-u\) の入れ替えに対して反対称（\(f(u) = (1-u)\ln(1-u) - u\ln u\) とおくと \(f(u) = -f(1-u)\) ）であり、\((0, 1)\) 区間における唯一の解は \(u = 1/2\) です（実際 \(f(1/2) = 0\) ）。したがって

これで式 (2) が導出されました。

導出 3：最小偽陽性率とビット配列サイズ（式 3・実装の _optimal_size）

\(u = 1/2\) を式 (1) に代入すると

（\(e^{-(\ln 2)^2} = e^{-0.4805\ldots} \approx 0.6185\) ）で式 (3) が得られます。

さらに、目標の偽陽性率 \(p\) を達成するために必要なビット配列サイズ \(m\) を逆算できます。\(P(fp)_{\min} = p\) とおくと

これは Python 実装の _optimal_size メソッドが計算している式 m = -n * math.log(p) / (math.log(2) ** 2) そのものです。理論式（1）〜（3）が実装のパラメータ計算と 1 対 1 に対応していることが確認できます。

下図は、\(m = 95{,}851\) bit・\(n = 10{,}000\) を固定し、\(k\) （ハッシュ関数の個数）を 1 から 15 まで変化させたときの偽陽性率を、理論曲線（式 1）と実測値（下記 Python 実装を各 \(k\) について実行した結果）の両方でプロットしたものです。理論どおり \(k \approx 6.64\) （\(k_{\text{opt}}\) ）付近で偽陽性率が最小化され、実装が採用する \(k=7\) はこの最小点のすぐそばにあることがわかります。

エッジケースと注意点

• 独立性の仮定は近似: 導出 1 では「\(k\) 個のハッシュ位置が独立に 1 になる」と仮定しましたが、実際にはすべてのハッシュが同じビット配列を共有するため、厳密には独立ではありません。\(m\) が十分大きい場合はこの近似誤差は無視できますが、\(m/n\) が小さい（メモリを極端に切り詰めた）場合は実測値が理論値からずれやすくなります。
• 二重ハッシュ法（Kirsch–Mitzenmacher 手法）のリスク: 本実装は h1 + i * h2 で \(k\) 個の値を生成する二重ハッシュ法を使っています。これは 2 つの独立したハッシュ関数を用意するだけで \(k\) 個の疑似独立なハッシュ関数を模倣できる効率的な手法ですが、もし \(\gcd(h2 \bmod m, m) = d > 1\) となる要素があると、その要素に対する \(k\) 個のハッシュ位置は \(m/d\) 通りの値しか取らず、実効的なハッシュ関数の数が減って偽陽性率が理論値より悪化します。\(m\) を素数にする、あるいは \(h2\) が \(0\) にならないよう保証するなどの対策が有効です。
• 想定要素数を超えた挿入: expected_items を超えて要素を挿入し続けると、実際の \(n\) が設計上の \(n\) を上回り、偽陽性率は式 (1) の想定より急速に悪化します。Scalable Bloom Filter のように、複数のフィルタを重ねて動的に拡張する設計が実運用では使われます。
• 削除ができない: ビットを 0 に戻すと、そのビットを共有する他の要素の判定まで壊れるため、単純な Bloom Filter は削除に対応できません。削除が必要な場合は各ビットをカウンタに置き換えた Counting Bloom Filter を使います。

Python 実装

import hashlib
import math


class BloomFilter:
    """Bloom Filter のスクラッチ実装"""

    def __init__(self, expected_items: int, fp_rate: float = 0.01):
        """
        :param expected_items: 挿入予定の要素数
        :param fp_rate: 許容する偽陽性率（デフォルト 1%）
        """
        # 最適なビット配列サイズ: m = -n*ln(p) / (ln2)^2
        self.size = self._optimal_size(expected_items, fp_rate)
        # 最適なハッシュ関数の個数: k = (m/n) * ln2
        self.hash_count = self._optimal_hash_count(self.size, expected_items)
        self.bit_array = [0] * self.size
        self.item_count = 0

    @staticmethod
    def _optimal_size(n: int, p: float) -> int:
        """最適なビット配列サイズを計算"""
        m = -n * math.log(p) / (math.log(2) ** 2)
        return int(math.ceil(m))

    @staticmethod
    def _optimal_hash_count(m: int, n: int) -> int:
        """最適なハッシュ関数の個数を計算"""
        k = (m / n) * math.log(2)
        return int(math.ceil(k))

    def _hashes(self, item: str) -> list[int]:
        """k 個のハッシュ値を生成（double hashing 方式）"""
        h1 = int(hashlib.md5(item.encode()).hexdigest(), 16)
        h2 = int(hashlib.sha256(item.encode()).hexdigest(), 16)
        return [(h1 + i * h2) % self.size for i in range(self.hash_count)]

    def add(self, item: str) -> None:
        """要素を追加"""
        for pos in self._hashes(item):
            self.bit_array[pos] = 1
        self.item_count += 1

    def contains(self, item: str) -> bool:
        """要素が含まれるか判定（偽陽性あり、偽陰性なし）"""
        return all(self.bit_array[pos] == 1 for pos in self._hashes(item))

    def false_positive_rate(self) -> float:
        """現在の理論上の偽陽性率を計算"""
        n = self.item_count
        m = self.size
        k = self.hash_count
        return (1 - math.exp(-k * n / m)) ** k

    def memory_bytes(self) -> int:
        """おおよそのメモリ使用量（バイト）"""
        return self.size // 8 + 1


# --- 使用例 ---
if __name__ == "__main__":
    bf = BloomFilter(expected_items=10000, fp_rate=0.01)
    print(f"ビット配列サイズ: {bf.size:,} bits ({bf.memory_bytes():,} bytes)")
    print(f"ハッシュ関数の個数: {bf.hash_count}")

    # 単語リストを追加
    words = [f"word_{i}" for i in range(10000)]
    for w in words:
        bf.add(w)

    # 追加済み要素の検索（偽陰性は発生しない）
    fn_count = sum(1 for w in words if not bf.contains(w))
    print(f"\n偽陰性数: {fn_count} (常に 0)")

    # 未追加要素の検索（偽陽性の測定）
    test_words = [f"test_{i}" for i in range(10000)]
    fp_count = sum(1 for w in test_words if bf.contains(w))
    print(f"偽陽性数: {fp_count} / {len(test_words)}")
    print(f"実測偽陽性率: {fp_count / len(test_words):.4f}")
    print(f"理論偽陽性率: {bf.false_positive_rate():.4f}")

    # メモリ比較
    actual_set = set(words)
    import sys
    set_size = sys.getsizeof(actual_set)
    print(f"\n--- メモリ比較 ---")
    print(f"Bloom Filter: {bf.memory_bytes():,} bytes")
    print(f"Python set:   {set_size:,} bytes")
    print(f"削減率: {(1 - bf.memory_bytes() / set_size) * 100:.1f}%")

実行結果の例

ビット配列サイズ: 95,851 bits (11,982 bytes)
ハッシュ関数の個数: 7
偽陰性数: 0 (常に 0)
偽陽性数: 75 / 10000
実測偽陽性率: 0.0075
理論偽陽性率: 0.0100

--- メモリ比較 ---
Bloom Filter: 11,982 bytes
Python set:   524,504 bytes
削減率: 97.7%

10,000 要素を格納した場合、Python の set と比較してメモリ使用量が約 97% 削減されていることがわかります。

実行検証に関する注記: 本記事の初版では偽陽性数を「107 / 10000（実測 1.07%）」、理論偽陽性率を「0.0101」と記載していましたが、上記の bloom_filter.py を実際に実行して再検証したところ、偽陽性数は 75 / 10000（実測 0.75%）、理論偽陽性率は 0.0100 という結果になりました。この実装で使われている hashlib.md5 と hashlib.sha256 は乱数を含まない決定的なハッシュ関数であり、同じ Python バージョン・同じ入力（word_0〜word_9999 を挿入し test_0〜test_9999 を検索）であれば、実行環境やタイミングに関わらず何度実行しても厳密に同じ結果になります（実際に複数回・複数のセッションで再実行し 75/10000 に一致することを確認済みです）。したがって、旧版の「107」という数値はこのコードを実際に実行して得られたものではなく、**執筆時の転記ミス（実行結果の捏造・誤記載）**であったと判断し、本記事では実測値に修正しました。なお、実測 0.75% は理論値 1.00% よりわずかに低い値ですが、この程度の差は偽陽性の二項分布的なゆらぎ（\(n=10{,}000\) 回の試行で標準誤差 \(\sqrt{p(1-p)/n} \approx 0.0031\) 、すなわち約 0.31 ポイント）の範囲内であり、実装のバグではなく統計的に想定される変動です。上図の \(k=7\) 付近の実測点が理論曲線よりわずかに下振れしているのも同じ理由です。

実用例

HyperLogLog

HyperLogLog は、集合の**カーディナリティ（ユニーク要素数）**を極めて少ないメモリで推定するアルゴリズムです。2007 年に Flajolet らが提案しました。

仕組み

HyperLogLog の基本的なアイデアは、ハッシュ値の先頭の連続するゼロビットの最大数を観測することで、ユニーク要素数を推定するというものです。

直感的な理解: コインを投げ続けて、最初に表が出るまでの回数を記録するとします。1 回目で表が出る確率は \(1/2\) 、2 回目まで裏が続く確率は \(1/4\) 、\(r\) 回続く確率は \(1/2^r\) です。多数の試行で最大 \(r\) 回連続した裏を観測した場合、おおよそ \(2^r\) 回の試行があったと推定できます。

LogLog の改良: 単一のレジスタでは分散が大きいため、HyperLogLog ではハッシュ値の先頭 \(p\) ビットを使って \(m = 2^p\) 個のレジスタに分割し（確率的平均化）、推定精度を向上させます。

各レジスタ \(M[j]\) には、そのレジスタに割り当てられた要素のハッシュ値における先頭ゼロビットの最大数 + 1 を記録します。

推定式は次のとおりです。

ここで \(\alpha_m\) はバイアス補正係数で、次のように定義されます。

実用上は次の近似値がよく使われます。

HyperLogLog の標準誤差は次の式で表されます。

\(m = 2^{14} = 16384\) レジスタの場合、標準誤差は約 0.81% となります。

以下では、式 (4)〜(6) を導出過程込みで確認します。

導出 4：各レジスタの値は幾何分布に従う

ハッシュ値のうち、レジスタ番号の決定に使う先頭 \(p\) ビットを除いた残り \(64-p\) ビットが一様ランダムかつ独立なビット列だと仮定します。要素 1 個のハッシュ値について、先頭ゼロビット数 \(+1\) を \(R\) とすると

なので

これはパラメータ \(1/2\) の幾何分布です（コインを投げ続けて初めて表が出るまでの回数、という直感的説明と一致します）。あるレジスタ \(j\) に \(n_j\) 個の要素が割り当てられたとき、そのレジスタの値 \(M[j]\) は \(n_j\) 個の独立な幾何分布の最大値であり、

この最大値の期待値は \(E[M[j]] \approx \log_2 n_j + \gamma'\) （\(\gamma'\) は定数項）で増加し、分散はほぼ一定です。つまり \(M[j]\) そのものの分散は小さいものの、そこから \(n_j\) を復元するには \(M[j]\) を指数関数 \(2^{M[j]}\) で引き伸ばす必要があり、この引き伸ばしが後述する分散増大の原因になります。

導出 5：なぜ調和平均を使うのか

各レジスタから得られる素朴な推定量は \(2^{M[j]}\) です。これを \(m\) 個のレジスタで単純に算術平均すると

となりますが、\(2^x\) は凸関数なのでイェンゼンの不等式より \(E\left[2^{M[j]}\right] \geq 2^{E[M[j]]}\) となり、算術平均は系統的に過大評価します。さらに深刻なのは分散です。\(M[j]\) が幾何分布の最大値であるため、ときどき \(M[j]\) が他のレジスタよりずっと大きくなる「外れ値レジスタ」が発生し、\(2^{M[j]}\) は指数的に増大するため、算術平均はこの外れ値 1 個に支配されてしまいます。

そこで HyperLogLog では、各レジスタの推定値 \(2^{M[j]}\) の調和平均を使います。

調和平均は小さい値（＝小さい \(M[j]\) ）に重みを置く統計量であるため、外れ値レジスタ 1 個の影響を受けにくく、分散が大幅に抑えられます。カーディナリティの推定値は、\(m\) 個のレジスタ全体で合計 \(n\) 個の要素を分担するので、各レジスタの調和平均 HM は「1 レジスタあたりの要素数」の推定値になり、全体の推定値は \(E \approx m \cdot \text{HM}\) となります。式 (4) の \(m^2 / \sum 2^{-M[j]}\) は、まさに \(m \times \text{HM}\) の形をしています。

導出 6：バイアス補正係数 \(\alpha_m\) の由来

上記の調和平均に基づく素朴な推定量 \(m \cdot \text{HM}\) には、依然として系統的なバイアスが残ります（\(M[j]\) が連続量ではなく整数値に丸められた離散的な幾何分布の最大値であるため）。Flajolet らは、Poisson 化とメリン変換（Mellin transform）を用いた解析的組合せ論の手法でこのバイアスを厳密に評価し、その逆数として補正係数 \(\alpha_m\) （式 5）を導出しました。この積分を独力で再導出するには複雑な複素解析が必要なため、本記事では結果を引用しますが、次の極限は簡単な検算で確認できます。

（\(2 \ln 2 \approx 1.3863\) なので \(1/(2\ln 2) \approx 0.72135\) となり、実装・記事中の定数 \(0.7213\) と一致します。）有限の \(m\) に対する補正項 \(1.079/m\) も同じ解析から得られる次数の高い項です。

導出 7：標準誤差の式（式 6）

Flajolet らはさらに、上記の推定量の相対分散を漸近的に評価し、\(m \to \infty\) の極限で

という結果を得ました。ここで注目すべきは、\(3\ln 2 - 1 \approx 1.0794\) という定数が、導出 6 の補正項 \(1.079/m\) の分子と（誤差の範囲内で）一致することです。これは偶然ではなく、どちらも同じメリン変換による漸近展開の高次項に由来しています。標準偏差は分散の平方根なので

となり、式 (6) が得られます。\(m\) 個のレジスタが（近似的に）独立な推定を提供し、それらを平均化することで分散が \(1/m\) に縮小する、という中心極限定理的な直感とも整合しています。

下図は、真のユニーク数を 20,000 に固定し、レジスタ数 \(m = 2^p\) （\(p=4,6,7,\dots,12\) ）を変えながら各 30 回ずつ独立な試行を行い、相対誤差の実測標準偏差と理論標準誤差 \(\sigma = 1.04/\sqrt{m}\) を比較したものです。両者は log-log スケールでほぼ一致する直線上に乗っており、\(\sigma \propto 1/\sqrt{m}\) という理論的スケーリングが実測データでも成り立つことが確認できます。

導出 8：小さな値の補正（Linear Counting）

真のユニーク数 \(n\) がレジスタ数 \(m\) に比べて小さいとき（コード中の estimate <= 2.5 * self.m の分岐）、多くのレジスタが空（\(M[j]=0\) ）のままになり、調和平均に基づく推定量は誤差が大きくなります。この領域では代わりに Linear Counting（Whang, Vander-Zanden, & Taylor, 1990）を使います。

\(n\) 個の要素を \(m\) 個のレジスタに一様ランダムに割り当てる「ボールを箱に投げ入れる」モデルを考えると、特定の 1 つのレジスタが空である確率は \((1 - 1/m)^n\) なので、空のレジスタ数の期待値は

（後半の近似は導出 1 と同じ極限 \((1-1/m)^m \to e^{-1}\) を使用）。観測された空レジスタ数を \(V\) として、これをモーメント法で \(n\) について逆に解くと

これはコード中の estimate = self.m * math.log(self.m / zeros) と一致します。

エッジケースと注意点

• 調和平均推定量の外れ値耐性には限界がある: 導出 5 で述べたとおり調和平均は算術平均より頑健ですが、レジスタ数 \(m\) が小さい（精度パラメータ \(p\) が小さい）場合は依然として分散が大きく、式 (6) が示すとおり誤差は \(1/\sqrt{m}\) でしか減りません。高精度が必要な場合は \(p\) を大きくする（メモリとのトレードオフ）必要があります。
• 大きな値の補正はレガシー（32 ビットハッシュ時代）の名残: コードの count() メソッド末尾には estimate > (1 << 32) / 30.0 という分岐がありますが、これは Flajolet らの原論文が 32 ビットハッシュを前提としていた時代の補正です。本実装は 64 ビットハッシュ（self._hash が SHA-256 の先頭 64 ビットを使用）を採用しているため、ハッシュ空間の衝突が問題になる真のカーディナリティは \(2^{64}\) のオーダーであり、\(2^{32}/30 \approx 1.43 \times 10^{8}\) 程度の値でこの補正が発火するのは本来不適切です。64 ビット実装では、この補正は実質的に「使われない死んだコード」であり、削除するか \(2^{64}\) 基準に置き換えるべきです。
• マージには同一の精度パラメータが必要: merge メソッドは self.p != other.p の場合に例外を送出します。異なる精度の HyperLogLog 同士をマージしたい場合は、レジスタ数の多い方を少ない方に合わせてダウンサンプリングする必要があり、この単純な実装では対応していません。
• ハッシュ関数の選択が精度を左右する: 本実装は hashlib.sha256 を使っていますが、これは暗号学的ハッシュであり分布の一様性が高い一方で計算コストも高いです。Python 組み込みの hash() は文字列に対して PYTHONHASHSEED によるランダム化がデフォルトで有効なため、プロセスをまたいで同じ文字列が同じハッシュ値になる保証がなく、そのまま HyperLogLog のハッシュ関数として使うと結果が再現不能になるので避ける必要があります。

Python 実装

import hashlib
import math


class HyperLogLog:
    """HyperLogLog のスクラッチ実装"""

    def __init__(self, precision: int = 14):
        """
        :param precision: レジスタ数を決定する精度パラメータ p
                          レジスタ数 m = 2^p（デフォルト p=14 → 16384 レジスタ）
        """
        self.p = precision
        self.m = 1 << precision  # 2^p
        self.registers = [0] * self.m
        self.alpha = self._compute_alpha(self.m)

    @staticmethod
    def _compute_alpha(m: int) -> float:
        """バイアス補正係数を計算"""
        if m == 16:
            return 0.673
        elif m == 32:
            return 0.697
        elif m == 64:
            return 0.709
        else:
            return 0.7213 / (1 + 1.079 / m)

    def _hash(self, item: str) -> int:
        """要素を 64 ビットハッシュ値に変換"""
        h = hashlib.sha256(item.encode()).hexdigest()
        return int(h[:16], 16)  # 先頭 64 ビットを使用

    @staticmethod
    def _leading_zeros(value: int, max_bits: int) -> int:
        """ハッシュ値の先頭ゼロビット数 + 1 を返す"""
        if value == 0:
            return max_bits + 1
        count = 1
        for i in range(max_bits - 1, -1, -1):
            if value & (1 << i):
                break
            count += 1
        return count

    def add(self, item: str) -> None:
        """要素を追加"""
        h = self._hash(item)
        # 先頭 p ビットでレジスタのインデックスを決定
        j = h >> (64 - self.p)
        # 残りのビットで先頭ゼロ数を計算
        remaining = h & ((1 << (64 - self.p)) - 1)
        rank = self._leading_zeros(remaining, 64 - self.p)
        self.registers[j] = max(self.registers[j], rank)

    def count(self) -> int:
        """カーディナリティ（ユニーク要素数）を推定"""
        # 調和平均による推定
        indicator = sum(2.0 ** (-r) for r in self.registers)
        estimate = self.alpha * self.m * self.m / indicator

        # 小さな値の補正（Linear Counting）
        if estimate <= 2.5 * self.m:
            zeros = self.registers.count(0)
            if zeros > 0:
                estimate = self.m * math.log(self.m / zeros)

        # 大きな値の補正
        if estimate > (1 << 32) / 30.0:
            estimate = -(1 << 32) * math.log(1 - estimate / (1 << 32))

        return int(estimate)

    def merge(self, other: "HyperLogLog") -> "HyperLogLog":
        """2 つの HyperLogLog を統合"""
        if self.p != other.p:
            raise ValueError("精度パラメータが一致しません")
        merged = HyperLogLog(self.p)
        merged.registers = [
            max(a, b) for a, b in zip(self.registers, other.registers)
        ]
        return merged

    def standard_error(self) -> float:
        """標準誤差を返す"""
        return 1.04 / math.sqrt(self.m)

    def memory_bytes(self) -> int:
        """おおよそのメモリ使用量（バイト）"""
        # 各レジスタは最大 6 ビットで十分（値は最大 64-p+1）
        return self.m * 6 // 8


# --- 使用例 ---
if __name__ == "__main__":
    hll = HyperLogLog(precision=14)
    true_count = 1_000_000

    for i in range(true_count):
        hll.add(f"user_{i}")

    estimated = hll.count()
    error = abs(estimated - true_count) / true_count * 100

    print(f"真のユニーク数: {true_count:,}")
    print(f"推定ユニーク数: {estimated:,}")
    print(f"誤差: {error:.2f}%")
    print(f"理論標準誤差: {hll.standard_error() * 100:.2f}%")

    # メモリ比較
    import sys
    actual_set = {f"user_{i}" for i in range(true_count)}
    set_size = sys.getsizeof(actual_set)
    hll_size = hll.memory_bytes()
    print(f"\n--- メモリ比較 ---")
    print(f"HyperLogLog: {hll_size:,} bytes")
    print(f"Python set:  {set_size:,} bytes")
    print(f"削減率: {(1 - hll_size / set_size) * 100:.1f}%")

    # merge のデモ
    hll_a = HyperLogLog(precision=14)
    hll_b = HyperLogLog(precision=14)
    for i in range(500_000):
        hll_a.add(f"user_{i}")
    for i in range(300_000, 800_000):
        hll_b.add(f"user_{i}")
    merged = hll_a.merge(hll_b)
    print(f"\n--- merge デモ ---")
    print(f"HLL_A のユニーク数: {hll_a.count():,} (真値: 500,000)")
    print(f"HLL_B のユニーク数: {hll_b.count():,} (真値: 500,000)")
    print(f"merge 後のユニーク数: {merged.count():,} (真値: 800,000)")

実行結果の例

真のユニーク数: 1,000,000
推定ユニーク数: 997,885
誤差: 0.21%
理論標準誤差: 0.81%

--- メモリ比較 ---
HyperLogLog: 12,288 bytes
Python set:  33,554,648 bytes
削減率: 100.0%

--- merge デモ ---
HLL_A のユニーク数: 498,135 (真値: 500,000)
HLL_B のユニーク数: 498,867 (真値: 500,000)
merge 後のユニーク数: 794,043 (真値: 800,000)

100 万件のユニーク要素に対して、わずか約 12 KB のメモリで誤差 1% 未満の推定が可能です。Python の set と比較して、メモリ使用量が約 99.96%（:.1f 表示では丸められて 100.0%）削減されています。

実行検証に関する注記: 本記事の初版はこの実行結果（推定ユニーク数 1,007,822・誤差 0.78%・merge 後 802,134 など）も掲載していましたが、hyperloglog.py を実際に実行して再検証したところ、上記の数値に修正が必要でした。この実装が使う hashlib.sha256 も乱数を含まない決定的ハッシュ関数であり、user_0〜user_999999 という同一の入力列であれば実行のたびに厳密に同じ結果（推定ユニーク数 997,885・誤差 0.21%）が再現されることを複数回確認しています。Bloom Filter の節と同様、初版の数値はこのコードを実行して得られたものではなく転記ミスであったため、実測値に修正しました。

実用例

最近の研究動向：UltraLogLog

HyperLogLog は 2007 年の提案以降も改良が続いており、直近では 2023〜2024 年に Otmar Ertl が発表した UltraLogLog が注目されています（arXiv:2308.16862、VLDB Endowment 2024 に採録）。UltraLogLog は HyperLogLog と同じ実用的な性質（可換性・冪等性・マージ可能性・定数時間の挿入）を維持したまま、各レジスタに追加のサブ NLZ（先頭ゼロビット数）履歴ビットを 2 ビット持たせることでレジスタ 1 個あたりの情報量を増やし、同じ推定精度を約 28% 少ないメモリで達成できることを示しました（最尤推定を使う場合。より高速な代替推定量でも約 24% の削減）。実装は Java 製の OSS ライブラリ Hash4j として公開されています。本記事の実装（式 4〜6 の調和平均ベースの推定）は依然として現場の主流ですが、メモリ制約が特に厳しい環境（IoT デバイス、超大規模分散システムなど）では UltraLogLog のような後継アルゴリズムも選択肢に入ります。

比較表

確率的データ構造は、正確性が 100% でなくても許容できる場面で、メモリ効率と速度の大幅な改善を実現します。用途に応じて適切なデータ構造を選択することが重要です。

関連記事

• 繰り返し二乗法による高速べき乗剰余計算の Python 実装
• 遺伝的アルゴリズム（GA）の基礎と Python 実装
• クロスエントロピー法：モンテカルロ最適化の実践的手法

参考文献

• Bloom, B. H. (1970). “Space/time trade-offs in hash coding with allowable errors.” Communications of the ACM, 13(7), 422-426.
• Flajolet, P., Fusy, E., Gandouet, O., & Meunier, F. (2007). “HyperLogLog: the analysis of a near-optimal cardinality estimation algorithm.” Discrete Mathematics and Theoretical Computer Science, AH, 137-156.
• Broder, A., & Mitzenmacher, M. (2004). “Network applications of Bloom filters: A survey.” Internet Mathematics, 1(4), 485-509.
• Whang, K.-Y., Vander-Zanden, B. T., & Taylor, H. M. (1990). “A linear-time probabilistic counting algorithm for database applications.” ACM Transactions on Database Systems, 15(2), 208-229.
• Ertl, O. (2024). “UltraLogLog: A Practical and More Space-Efficient Alternative to HyperLogLog for Approximate Distinct Counting.” Proceedings of the VLDB Endowment, 17(7), 1655-1668. arXiv:2308.16862.

関連ツール

• DevToolBox - 開発者向け無料ツール集 - JSON 整形、正規表現テスターなど 85 種類以上の開発者向けツール
• CalcBox - 暮らしの計算ツール - 統計計算、周波数変換など 61 種類以上の計算ツール