現場で直面した課題：メモリの壁への衝突
数年前、私たちのバックエンドチームは、毎秒約12万リクエストを処理する広告ネットワーク向けのクリックストリームイベント取り込みパイプラインを構築しました。その際、受信するすべてのリクエストに対して、ミリ秒未満で完了する3つの操作が必要でした。
- ユーザーIDがアクティブなキャンペーンをすでに閲覧済みかどうかの確認（メンバーシップ判定）
- 媒体ドメインごとの日次ユニークビジター数の追跡（カーディナリティ推定）
- DDoS攻撃を緩和するためのIPアドレスごとのリクエスト頻度カウント（頻度推定）
当初は標準的なPythonのコレクションを使用していました。ユーザーIDをsetに格納し、IPの出現頻度をdictで追跡していました。ステージング環境での小規模なサンプリングテストでは、すべてがスムーズに動作していました。しかし、本番環境への投入で問題が発生します。
パイプラインが5,000万件のユニークレコードを取り込んだ時点で、ワーカプロセスは20分足らずで16GBのRAMを消費し尽くしました。LinuxのOOM Killerによってワーカーが強制終了され、取り込みキューには未読のKafkaメッセージが400万件滞留し、パイプライン全体が完全に停止してしまったのです。
根本原因の分析：大規模データで厳密なコレクションが破綻する理由
標準的なデータ構造が高トラフィックなストリーミング処理で破綻する背景には、CPythonの3つの根本的な制約があります。
- 線形空間計算量 O(N)：厳密な検索を行うには、すべての個別要素をRAM上に保持する必要があります。1億個の生の64ビット整数を保持するだけでも、純粋なバイナリデータとして約800MBを消費します。
- PyObjectのオーバーヘッド：CPythonはすべての整数や文字列をリッチな
PyObjectヘッダーでラップします。そのため、単一の64ビット整数であっても8バイトではなく28バイトを消費します。さらにハッシュテーブルのポインタ配列、リサイズ用バッファ、バケットのオーバーヘッドなどを考慮すると、setの1エントリは80–100バイトにまで膨れ上がります。結果として、1億個の整数がヒープ領域で約9GBも消費することになります。 - ガベージコレクションのスラッシング：無数の細かなヒープ割り当てによって激しいメモリの断片化が発生します。Pythonのガベージコレクタが巡回スイープ処理を実行するたびにワーカースレッドが停止し、サービス全体のレイテンシスパイクを引き起こします。
代替アプローチの比較検討
インメモリのPythonコレクションが限界を迎えた場合、一般的には次の3つの選択肢が検討されます。
- 垂直スケーリング（インスタンスの増強）：128GBや256GBのAWSインスタンスを投入することで一時的には解決しますが、コストが急増します。複数のワーカーで10億件の識別子を保持する場合、クラウドアカウントのメモリ費用だけで月額数千ドルに達することもあります。
- 外部キーバリューストア（Redis、RocksDB、Cassandra）：状態を外部にオフロードすることでPythonプロセスを軽量に保てます。しかし、ネットワークホップが発生することで検索ごとに1–3ミリ秒の遅延が加わります。秒間8万クエリの規模では、ネットワークのシリアライズやコネクションプールの競合が新たなボトルネックとなります。
- 確率的データ構造（Bloom Filter、HyperLogLog、Count-Min Sketch）：これらのアルゴリズムは、数学的な厳密性を犠牲にする代わりに、厳密かつ決定論的なメモリ上限を実現します。「1%の偽陽性率」や「±1.04%のカーディナリティ誤差」といった許容誤差を設定することで、O(1)の検索性能を維持しながら1MB未満の固定メモリフットプリントを実現できます。
実践的な実装：主要な3つの確率的データ構造
パイプラインを確率的データ構造へ移行したことで、ワーカーインスタンスあたりのメモリ使用量は16GBから約18MBへと劇的に削減されました。これにより、サービスは1件のイベントもドロップすることなくトラフィックスパイクを乗り切ることができました。
1. メンバーシップ判定のためのBloom Filter
Bloom Filter（ブルームフィルタ）は、ある要素が集合に「絶対に存在しない」か「おそらく存在する」かを判定します。サイズmのビット配列とk個の独立したハッシュ関数を使用します。偽陰性（False Negative）は原理的に発生せず、偽陽性（False Positive）の確率は数学的に制御可能です。
import math
import mmh3
from bitarray import bitarray
class BloomFilter:
def __init__(self, expected_elements: int, false_positive_rate: float):
self.expected_elements = expected_elements
self.false_positive_rate = false_positive_rate
# 最適なビット配列サイズ: m = -(n * ln(p)) / (ln(2)^2)
self.size = int(- (expected_elements * math.log(false_positive_rate)) / (math.log(2) ** 2))
# 最適なハッシュ関数の数: k = (m / n) * ln(2)
self.hash_count = int((self.size / expected_elements) * math.log(2))
self.bit_array = bitarray(self.size)
self.bit_array.setall(0)
def add(self, item: str):
for seed in range(self.hash_count):
index = mmh3.hash(item, seed) % self.size
self.bit_array[index] = 1
def contains(self, item: str) -> bool:
for seed in range(self.hash_count):
index = mmh3.hash(item, seed) % self.size
if not self.bit_array[index]:
return False # 確実に存在しない
return True # 存在する可能性が高い
# ベンチマーク: 偽陽性率1%で100万アイテムを格納した場合のメモリは約1.14 MB
bf = BloomFilter(expected_elements=1_000_000, false_positive_rate=0.01)
bf.add("user_8492048")
print(bf.contains("user_8492048")) # True
print(bf.contains("user_9999999")) # False
2. カーディナリティ推定のためのHyperLogLog (HLL)
数十億規模のイベントからユニークビジター数をカウントする処理は、従来非常に高コストでした。HyperLogLogは、ハッシュ値の先頭に連続するゼロ（Leading Zeros）の最大長を追跡することで、全体のカーディナリティを推定します。$2^{14} = 16,384$個のレジスタを使用した場合、HLLはわずか12KBのRAM消費で約1.04%の誤差率を維持します。
import hashlib
import math
class HyperLogLog:
def __init__(self, p: int = 14):
self.p = p # 精度パラメータ
self.m = 1 << p # レジスタ数（p=14の場合は16,384）
self.registers = [0] * self.m
self.alpha_m = self._get_alpha(self.m)
def _get_alpha(self, m: int) -> float:
if m == 16:
return 0.673
elif m == 32:
return 0.697
elif m == 64:
return 0.709
return 0.7213 / (1 + 1.079 / m)
def _count_leading_zeros(self, value: int, bits: int) -> int:
if value == 0:
return bits
count = 0
while (value & (1 << (bits - 1))) == 0:
count += 1
value <<= 1
return count + 1
def add(self, item: str):
digest = hashlib.sha256(item.encode('utf-8')).digest()
hash_val = int.from_bytes(digest[:8], byteorder='big')
# 先頭pビットでレジスタのインデックスを決定
register_index = hash_val >> (64 - self.p)
# 残りのビットで先行ゼロの連続数を評価
w = hash_val & ((1 << (64 - self.p)) - 1)
leading_zeros = self._count_leading_zeros(w, 64 - self.p)
self.registers[register_index] = max(self.registers[register_index], leading_zeros)
def count(self) -> int:
# すべてのレジスタの調和平均を計算
raw_estimate = self.alpha_m * (self.m ** 2) / sum(2.0 ** (-r) for r in self.registers)
# スパースなレジスタに対するリニアカウンティング補正
if raw_estimate <= 2.5 * self.m:
zeros = self.registers.count(0)
if zeros != 0:
return int(self.m * math.log(self.m / zeros))
return int(raw_estimate)
# ベンチマーク: 10万個の個別アイテムを12 KBで追跡
hll = HyperLogLog(p=14)
for i in range(100_000):
hll.add(f"client_id_{i}")
print(f"推定カーディナリティ: {hll.count()}") # 通常99,400〜100,600程度
3. 頻度推定のためのCount-Min Sketch
Count-Min Sketchは、2次元グリッド状のカウンタとして動作します。実際のカウント値を過小評価することなく、高速なデータストリームにおける出現頻度を推定します。過大評価される誤差は、確率1 - deltaのもとで指定した許容範囲epsilon内に収まります。
import math
import mmh3
class CountMinSketch:
def __init__(self, epsilon: float = 0.001, delta: float = 0.01):
# 幅（Width）で誤差の大きさを制御、深さ（Depth）で誤差発生確率を制御
self.width = int(math.ceil(math.e / epsilon)) # 約2,719列
self.depth = int(math.ceil(math.log(1.0 / delta))) # 約5行
self.table = [[0] * self.width for _ in range(self.depth)]
def add(self, item: str, count: int = 1):
for row in range(self.depth):
col = mmh3.hash(item, row) % self.width
self.table[row][col] += count
def estimate(self, item: str) -> int:
# 実際の出現頻度は、マッピングされた全セルの最小値以下となる
return min(
self.table[row][mmh3.hash(item, row) % self.width]
for row in range(self.depth)
)
# ベンチマーク: 2719x5の整数行列で60 KB未満のRAMを消費
cms = CountMinSketch(epsilon=0.001, delta=0.01)
cms.add("192.168.1.50", count=450)
cms.add("10.0.0.1", count=12)
print(f"推定リクエスト数: {cms.estimate('192.168.1.50')}") # 450
本番運用のガイドライン
- Pythonのハッシュループ処理のオーバーヘッドに注意：ペイロードごとに5–7回のハッシュを計算する純粋なPythonループは、CPUのボトルネックになる可能性があります。ワーカーあたり秒間20,000件を超えるスループットが必要な場合は、
pyprobablesやdatasketchなどのC拡張ライブラリ、あるいはネイティブのRedisモジュール（BF.ADD、PFADD、CMS.INCRBY）への切り替えを検討してください。 - 事前のサイジング：標準的なBloom Filterは、一度割り当てると動的に拡張できません。予想されるピークトラフィックの1.5倍から2倍のサイズで見積もるか、拡張可能な配列を連結するScalable Bloom Filterを実装してください。
- データベース層への事前フィルタリングの導入：PostgresやDynamoDBへのクエリ実行前に、メモリ上に500KB程度のBloom Filterを配置します。フィルタが
Falseを返した場合はクエリを完全にスキップします。ユーザー新規登録の重複チェックやコールドルックアップなど、キャッシュミスが多いシナリオでは、不要な読み取りI/Oを最大90%削減できます。
