Bloom Filter With Double Hashing
Size a Bloom filter from an expected item count and false-positive rate, then add and test members using double hashing.
Code
from __future__ import annotations
import hashlib
import math
class BloomFilter:
def __init__(
self,
expected_items: int,
false_positive_rate: float = 0.01,
) -> None:
if (
not isinstance(expected_items, int)
or isinstance(expected_items, bool)
or expected_items < 1
):
raise ValueError("expected_items must be a positive integer")
if (
not isinstance(false_positive_rate, (int, float))
or isinstance(false_positive_rate, bool)
or not 0 < false_positive_rate < 1
):
raise ValueError("false_positive_rate must be between 0 and 1")
self.expected_items = expected_items
self.false_positive_rate = float(false_positive_rate)
self.size = max(
1,
math.ceil(
-(expected_items * math.log(self.false_positive_rate))
/ (math.log(2) ** 2)
),
)
self.hash_count = max(
1,
round((self.size / expected_items) * math.log(2)),
)
self._bits = bytearray((self.size + 7) // 8)
self.added = 0
@staticmethod
def _require_text(value: str) -> str:
if not isinstance(value, str):
raise TypeError("value must be a string")
if not value:
raise ValueError("value must not be empty")
return value
def _positions(self, value: str):
digest = hashlib.sha256(value.encode("utf-8")).digest()
first = int.from_bytes(digest[:8], "big")
second = int.from_bytes(digest[8:16], "big") or 1
for index in range(self.hash_count):
yield (first + index * second) % self.size
def add(self, value: str) -> None:
for position in self._positions(self._require_text(value)):
self._bits[position // 8] |= 1 << (position % 8)
self.added += 1
def __contains__(self, value: str) -> bool:
text = self._require_text(value)
return all(
self._bits[position // 8] & (1 << (position % 8))
for position in self._positions(text)
)
@property
def fill_ratio(self) -> float:
bits_set = sum(byte.bit_count() for byte in self._bits)
return bits_set / self.size
def main() -> None:
bloom = BloomFilter(expected_items=1000, false_positive_rate=0.01)
for word in ["alpha", "beta", "gamma", "delta"]:
bloom.add(word)
print(all(word in bloom for word in ["alpha", "beta", "gamma", "delta"]))
print("epsilon" in bloom)
print(bloom.size, bloom.hash_count, round(bloom.fill_ratio, 4))
try:
BloomFilter(0)
except ValueError as error:
print(error)
if __name__ == "__main__":
main()Report a problem with this exampleOpens a pre-filled issue on GitHub.