Problem 92 · Data Engineering & Streaming

Count-Min Sketch

MediumClassic DSA & systems

01Problem

Implement a Count-Min Sketch supporting add(item, count), estimate(item), and merge, sized from error epsilon and failure probability delta.

02Key insight

d hash rows of width w; add increments one counter per row and estimate takes the min. Estimates never undercount, and overcount by at most eps*N with probability 1-delta when w=ceil(e/eps), d=ceil(ln(1/delta)). Sketches with equal shape and seeds merge by elementwise addition.

03Solution

solution.pypython34 lines

import hashlib
import math
from typing import Hashable, Iterator


class CountMinSketch:
    def __init__(self, width: int, depth: int, seed: int = 0) -> None:
        self.width, self.depth = width, depth
        self.table = [[0] * width for _ in range(depth)]
        self._keys = [f"cms:{seed}:{i}".encode() for i in range(depth)]

    @classmethod
    def from_error(cls, epsilon: float, delta: float, seed: int = 0) -> "CountMinSketch":
        return cls(math.ceil(math.e / epsilon), math.ceil(math.log(1 / delta)), seed)

    def _columns(self, item: Hashable) -> Iterator[int]:
        data = repr(item).encode()
        for key in self._keys:  # independent keyed hash per row
            digest = hashlib.blake2b(data, digest_size=8, key=key).digest()
            yield int.from_bytes(digest, "big") % self.width

    def add(self, item: Hashable, count: int = 1) -> None:
        for row, col in enumerate(self._columns(item)):
            self.table[row][col] += count

    def estimate(self, item: Hashable) -> int:
        return min(self.table[row][col] for row, col in enumerate(self._columns(item)))

    def merge(self, other: "CountMinSketch") -> None:
        if (self.width, self.depth, self._keys) != (other.width, other.depth, other._keys):
            raise ValueError("incompatible sketches")
        for mine, theirs in zip(self.table, other.table):
            for i, v in enumerate(theirs):
                mine[i] += v

Try it first.

TimeO(d) per add/estimate

SpaceO(w*d)

tests.py · 3 passing

tests.pypython13 lines

import random
from collections import Counter
rng = random.Random(7)
stream = [int(rng.paretovariate(1.2)) % 1000 for _ in range(20000)]
truth = Counter(stream)
a, b = CountMinSketch.from_error(0.002, 0.01), CountMinSketch.from_error(0.002, 0.01)
for i, x in enumerate(stream):
    (a if i % 2 else b).add(x)
a.merge(b)
eps_n = 0.002 * len(stream)
assert all(a.estimate(k) >= v for k, v in truth.items())          # never undercounts
assert sum(a.estimate(k) - v > eps_n for k, v in truth.items()) <= 0.02 * len(truth)
assert a.estimate("never-seen") <= eps_n

04Staff-level follow-up

Low-frequency items get large relative error - how do you improve it? Use conservative update (only raise counters that equal the current min), or pair the sketch with a small exact heap of heavy hitters; CMS is for heavy-hitter-ish point queries, not the long tail.