Chapter 29: Probabilistic Data Structures

1. Why Probabilistic

Probabilistic data structures trade perfect accuracy for massive improvements in space and time complexity. They provide approximate answers with bounded, quantifiable error rates.

Key Properties

2. Bloom Filter

A space-efficient probabilistic data structure for set membership testing. Uses a bit array and multiple hash functions.

Structure

Operations

class BloomFilter:
    def __init__(self, m, k):
        self.m = m              # bit array size
        self.k = k              # number of hash functions
        self.bits = [0] * m

    def add(self, item):
        for i in range(self.k):
            index = hash_i(item, i) % self.m
            self.bits[index] = 1

    def contains(self, item):
        for i in range(self.k):
            index = hash_i(item, i) % self.m
            if self.bits[index] == 0:
                return False    # Definitely not in set
        return True             # Probably in set (might be false positive)

False Positive Rate

Parameter Selection

3. Bloom Filter Applications

Cassandra SSTable Bloom Filters

// Query: SELECT * FROM users WHERE user_id = 12345
// Without Bloom filter: Check all 10 SSTables on disk
// With Bloom filter:
//   - 8 SSTables return "definitely not present" (skip disk I/O)
//   - 2 SSTables return "maybe present" (read these)
//   Result: 80% reduction in disk reads

// Configuration
CREATE TABLE users (
    user_id bigint PRIMARY KEY,
    name text
) WITH bloom_filter_fp_chance = 0.01;  // 1% FPR

Web Crawler Architecture

4. Counting Bloom Filter

Extension of Bloom filter that supports deletion by replacing each bit with a small counter (typically 4 bits).

Structure Comparison

Implementation

class CountingBloomFilter:
    def __init__(self, m, k, counter_bits=4):
        self.m = m
        self.k = k
        self.max_count = (1 << counter_bits) - 1  # 15 for 4-bit
        self.counters = [0] * m

    def add(self, item):
        for i in range(self.k):
            index = hash_i(item, i) % self.m
            if self.counters[index] < self.max_count:
                self.counters[index] += 1

    def remove(self, item):
        # First verify item is present
        if not self.contains(item):
            return False

        for i in range(self.k):
            index = hash_i(item, i) % self.m
            if self.counters[index] > 0:
                self.counters[index] -= 1
        return True

    def contains(self, item):
        for i in range(self.k):
            index = hash_i(item, i) % self.m
            if self.counters[index] == 0:
                return False
        return True

Counter Overflow Problem

5. Count-Min Sketch

Probabilistic data structure for frequency estimation. Uses a 2D array with multiple hash functions to estimate item counts with over-estimation guarantee.

Structure

Algorithm

class CountMinSketch:
    def __init__(self, width, depth):
        self.w = width          # w columns
        self.d = depth          # d rows (hash functions)
        self.table = [[0] * width for _ in range(depth)]

    def add(self, item, count=1):
        for i in range(self.d):
            index = hash_i(item, i) % self.w
            self.table[i][index] += count

    def estimate(self, item):
        estimates = []
        for i in range(self.d):
            index = hash_i(item, i) % self.w
            estimates.append(self.table[i][index])
        return min(estimates)   # Return minimum to reduce over-estimation

Error Guarantees

Parameter Selection

6. Count-Min Sketch Applications

Heavy Hitters Algorithm

class HeavyHitters:
    def __init__(self, k, epsilon, delta):
        self.k = k
        self.cms = CountMinSketch(width=int(2.718/epsilon), depth=int(math.log(1/delta)))
        self.heap = []  # Min-heap of (count, item)

    def add(self, item):
        self.cms.add(item)
        count = self.cms.estimate(item)

        # If item in heap, update it
        # If heap not full or count > min, add to heap
        if len(self.heap) < self.k:
            heapq.heappush(self.heap, (count, item))
        elif count > self.heap[0][0]:
            heapq.heapreplace(self.heap, (count, item))

    def get_top_k(self):
        return sorted(self.heap, reverse=True)

# Example: Find top 100 URLs with epsilon=0.001, delta=0.01
# Space: ~53 KB for CMS + ~3 KB for heap = 56 KB
# vs HashMap tracking all URLs: potentially GBs

Rate Limiting with CMS

7. HyperLogLog

Probabilistic algorithm for estimating cardinality (count of distinct elements) with remarkable space efficiency: 12 KB for 0.8% error on billions of items.

Core Intuition

Algorithm

Implementation

class HyperLogLog:
    def __init__(self, precision=14):
        self.p = precision              # Number of bits for bucket index
        self.m = 1 << precision         # Number of buckets (2^p)
        self.M = [0] * self.m           # Max leading zeros per bucket

        # Alpha constant for bias correction
        if self.m >= 128:
            self.alpha = 0.7213 / (1 + 1.079 / self.m)
        elif self.m >= 64:
            self.alpha = 0.709
        elif self.m >= 32:
            self.alpha = 0.697
        elif self.m >= 16:
            self.alpha = 0.673

    def add(self, item):
        # Hash to 64 bits
        h = hash64(item)

        # First p bits determine bucket
        j = h & ((1 << self.p) - 1)

        # Remaining bits: count leading zeros + 1
        w = h >> self.p
        rho = leading_zeros(w) + 1

        # Update max for this bucket
        self.M[j] = max(self.M[j], rho)

    def cardinality(self):
        # Harmonic mean
        raw_estimate = self.alpha * (self.m ** 2) / sum(2 ** (-x) for x in self.M)

        # Small range correction
        if raw_estimate <= 2.5 * self.m:
            zeros = self.M.count(0)
            if zeros != 0:
                return self.m * math.log(self.m / zeros)

        # Large range correction
        if raw_estimate > (1/30) * (2 ** 32):
            return -2 ** 32 * math.log(1 - raw_estimate / (2 ** 32))

        return raw_estimate

Precision vs Space Tradeoff

Merging HyperLogLogs

def merge(hll1, hll2):
    # HLLs must have same precision
    assert hll1.p == hll2.p

    result = HyperLogLog(hll1.p)
    for i in range(hll1.m):
        result.M[i] = max(hll1.M[i], hll2.M[i])

    return result

# Example: Union cardinality across shards
shard_hlls = [shard.get_hll() for shard in shards]
global_hll = reduce(merge, shard_hlls)
total_uniques = global_hll.cardinality()

8. HyperLogLog Applications

Redis HyperLogLog Commands

# Add elements
PFADD unique_visitors user:123 user:456 user:789

# Get cardinality estimate
PFCOUNT unique_visitors
# Returns: 3

# Merge multiple HLLs
PFADD page:home user:123 user:456
PFADD page:about user:456 user:789
PFMERGE total_visitors page:home page:about
PFCOUNT total_visitors
# Returns: 3 (union of both sets)

# Memory usage
MEMORY USAGE unique_visitors
# Returns: ~12KB regardless of cardinality

Multi-Dimensional Cardinality

# Track unique visitors per (country, device) dimension
# Without HLL: Need HashSet per dimension combo = 200 countries × 10 devices = 2000 sets
# With HLL: 12 KB × 2000 = 24 MB for all dimensions

for event in stream:
    key = f"uniques:{event.country}:{event.device}:{date}"
    redis.pfadd(key, event.user_id)

# Query: Unique mobile users in US on 2026-05-12
count = redis.pfcount("uniques:US:mobile:2026-05-12")

# Query: Unique users across all devices in France
keys = [f"uniques:FR:{device}:{date}" for device in devices]
redis.pfmerge("temp:query", *keys)
count = redis.pfcount("temp:query")

Real-World Performance

9. t-digest

Probabilistic data structure for estimating percentiles and quantiles from streaming data with bounded error and mergeability for distributed systems.

Core Concept

Why Non-Uniform Clustering

Implementation Sketch

class TDigest:
    def __init__(self, compression=100):
        self.compression = compression  # δ parameter
        self.centroids = []             # List of (mean, weight)
        self.count = 0

    def add(self, value, weight=1):
        self.count += weight

        # Find nearest centroid
        nearest = self._find_nearest(value)

        # Compute capacity using scaling function
        q = self._quantile_of_centroid(nearest)
        capacity = self._capacity_at_quantile(q)

        if nearest.weight + weight <= capacity:
            # Merge into existing centroid
            nearest.merge(value, weight)
        else:
            # Create new centroid
            self.centroids.append(Centroid(value, weight))

        # Periodically compress
        if len(self.centroids) > self.compression * 3:
            self._compress()

    def quantile(self, q):
        # q in [0, 1]
        target = q * self.count
        sum_weight = 0

        for i, c in enumerate(self.centroids):
            sum_weight += c.weight
            if sum_weight >= target:
                # Interpolate between centroids
                if i == 0:
                    return c.mean
                prev = self.centroids[i-1]
                delta = (target - (sum_weight - c.weight)) / c.weight
                return prev.mean + delta * (c.mean - prev.mean)

        return self.centroids[-1].mean

    def _capacity_at_quantile(self, q):
        # Scaling function k(q,δ) determines centroid capacity
        # More capacity near median, less at tails
        return 4 * self.count * q * (1 - q) / self.compression

    def merge(self, other):
        # Combine centroids from both digests
        self.centroids.extend(other.centroids)
        self.count += other.count
        self._compress()

Error Guarantees

Applications

10. Cuckoo Filter

Probabilistic data structure for set membership that supports deletion, offers better space efficiency than Bloom filters at low false positive rates, and provides faster lookups.

Key Differences from Bloom Filter

Structure

Algorithm

class CuckooFilter:
    def __init__(self, capacity, bucket_size=4, fingerprint_size=8):
        self.m = capacity // bucket_size    # Number of buckets
        self.b = bucket_size                # Entries per bucket
        self.f = fingerprint_size           # Fingerprint bits
        self.buckets = [[None]*self.b for _ in range(self.m)]

    def _fingerprint(self, item):
        return hash(item) % (1 << self.f)

    def _index(self, item):
        return hash(item) % self.m

    def _alt_index(self, index, fingerprint):
        return (index ^ hash(fingerprint)) % self.m

    def insert(self, item):
        f = self._fingerprint(item)
        i1 = self._index(item)
        i2 = self._alt_index(i1, f)

        # Try to insert in i1 or i2
        if self._insert_to_bucket(i1, f):
            return True
        if self._insert_to_bucket(i2, f):
            return True

        # Both buckets full: relocate existing items (cuckoo hashing)
        i = random.choice([i1, i2])
        for _ in range(500):  # Max kicks
            f, self.buckets[i][random.randint(0, self.b-1)] = \
                self.buckets[i][random.randint(0, self.b-1)], f
            i = self._alt_index(i, f)
            if self._insert_to_bucket(i, f):
                return True

        return False  # Filter full

    def contains(self, item):
        f = self._fingerprint(item)
        i1 = self._index(item)
        i2 = self._alt_index(i1, f)
        return f in self.buckets[i1] or f in self.buckets[i2]

    def delete(self, item):
        f = self._fingerprint(item)
        i1 = self._index(item)
        i2 = self._alt_index(i1, f)

        if f in self.buckets[i1]:
            self.buckets[i1].remove(f)
            return True
        if f in self.buckets[i2]:
            self.buckets[i2].remove(f)
            return True
        return False

Performance Comparison

11. Comparison Matrix

Detailed Feature Matrix

Error Characteristics

12. When to Use

Decision Tree by Problem Type

System Design Patterns

Common Combinations

Implementation Checklist