Data-Stream Algorithms

Almost every algorithm in this course so far has assumed the same machine: the input sits in memory, and any element is as cheap to touch as any other. That is the RAM model, and it underwrites our analyses of hash tables, balanced trees, and the rest. External sorting was our first crack in that assumption: the data was too big for RAM, so we counted block transfers to a disk we could re-read at will. This lesson drops even that assumption. In the streaming model, the input flows past once, item by item, faster and larger than we can store. We may keep only a tiny working set, and once an item slides by we may never see it again. The question is what we can still compute.

Three models, three currencies

It helps to put the models side by side. Each makes a different assumption about where the data lives and how often we may touch it, and each counts a different cost.

With so little memory we cannot, in general, answer exactly. The obstruction is information-theoretic. An algorithm with $s$ bits of memory has at most $2^s$ distinguishable states, so if two different prefixes of the stream drive it into the same state, its future behavior on any common suffix is identical. For a question like were all items distinct? there are more prefixes demanding different future behavior than a sublinear memory has states: two different $\frac{n}{2}$-element prefixes must be distinguished, because an adversary can extend each with an element that appears in one but not the other. Counting those prefixes forces $s=\Omega (n)$ bits for an exact answer, and a similar argument yields the same bound for the exact median.

So streaming algorithms return approximate or probabilistic answers, accepting a small, tunable error in exchange for space that is exponentially smaller than the stream. This is the same trade a Bloom filter makes for set membership, and, as there, the tools are hashing and randomized analysis.¹

The rest of the lesson is five techniques, each a small variation on that trade-off: sample the stream, count it, fingerprint its frequencies, find its heavy hitters, and estimate how many distinct things it contained.

Reservoir sampling: a uniform sample of unknown length

Problem. A stream of $n$ items flows past, with $n$ unknown until the end. Keep a uniform random sample of $k$ of them, using only $O(k)$ space, so that at every moment the $k$ stored items are a uniform random $k$-subset of everything seen so far.

The naive approach, store everything and sample at the end, needs $O(n)$ space and a known $n$. Reservoir sampling removes both costs with one idea: keep a reservoir of the first $k$ items, then for each later item decide on the spot whether it displaces one of them.

Idea. When the $i$-th item arrives ($i>k$), admit it into the reservoir with probability $k/i$; if admitted, it evicts a uniformly random current occupant. In a uniform size-$k$ sample of $i$ items, each item appears with probability exactly $k/i$, and that is the probability we give it.

One subtlety in the pseudocode: a single random draw $j\in [1,i]$ decides both questions at once. The event $j\le k$ has probability exactly $k/i$, which settles admission, and conditioned on admission $j$ is uniform over $[1,k]$, which settles who gets evicted. One random number per item, $O(1)$ work per item, $O(k)$ space total.

A worked trace. Run the algorithm with $k=2$ on the stream $a,b,c,d,e$. The first two items fill the reservoir, so after $i=2$ it holds $\{a,b\}$. Then the coin flips begin.

• Item $c$ ($i=3$). Draw $j\in [1,3]$; admission needs $j\le 2$, which happens with probability $2/3$. Suppose $j=1$: $c$ is admitted and evicts the occupant of slot $1$. Reservoir: $\{c,b\}$.
• Item $d$ ($i=4$). Admission probability $2/4=1/2$. Suppose $j=4$: the draw exceeds $k$, so $d$ is rejected and discarded forever. Reservoir unchanged: $\{c,b\}$.
• Item $e$ ($i=5$). Admission probability $2/5$. Suppose $j=2$: $e$ is admitted and evicts slot $2$. Reservoir: $\{c,e\}$.

The run above is one sample path; the theorem below says that averaging over all coin flips, every one of the $\left(\genfrac{}{}{0ex}{}{5}{2}\right)=10$ pairs is equally likely, and each individual item ends up retained with probability exactly $2/5$. This local rule produces a globally uniform sample at every prefix, not just at the end: stop the stream after any $i$, and the reservoir is a uniform $k$-subset of the first $i$ items.

Check the theorem against the trace: item $c$ arrived at $i=3$, so it should be retained with probability $\frac{2}{3}\cdot \frac{3}{4}\cdot \frac{4}{5}=\frac{2}{5}$, admission times two survivals. Item $a$, present from the start, must survive the arrivals at $i=3,4,5$, each of which evicts it with probability $\frac{1}{i}$: $\left(1-\frac{1}{3}\right)\left(1-\frac{1}{4}\right)\left(1-\frac{1}{5}\right)=\frac{2}{3}\cdot \frac{3}{4}\cdot \frac{4}{5}=\frac{2}{5}$ as well. Every path through the coin flips is different; the retention probability is not.

Edge cases and variants. If the stream ends with $n<k$ items, the reservoir simply holds all of them, which is the only correct answer. The special case $k=1$ is worth memorizing on its own: keep one item, and replace it with the $i$-th arrival with probability $1/i$. That is the entire solution to Linked List Random Node, a uniform pick from a list of unknown length in one pass and $O(1)$ space. For very long streams where the random-number generator is the bottleneck, the admission probability $k/i$ shrinks, so most draws are rejections; one can instead draw, in $O(1)$ time, the number of items to skip before the next admission, and fast-forward. And when items carry weights $w_i$ and the sample should favor heavy items, a weighted variant assigns each item the key $u_i^{1/w_i}$ for a uniform $u_i\in (0,1)$ and keeps the $k$ largest keys in a small heap; unweighted reservoir sampling is the special case where all weights are $1$. (The LeetCode problem Random Pick with Weight is the offline cousin: with all weights in memory, prefix sums and a binary search do the job.)

Reservoir sampling underlies A/B test logging, random log extraction, and fair sampling anywhere the choice is from a sequence whose length is learned only at the end.

import random
from collections.abc import Iterable
from typing import Generic, Optional, TypeVar

Item = TypeVar("Item")

class ReservoirSampler(Generic[Item]):
  """
    A streaming uniform sampler keeping at most `capacity` items.\n
    Feed items one at a time with `offer`; `sample` returns the current\n
    reservoir. An optional `seed` makes the sampling reproducible.\n
  """

  def __init__(self, capacity: int, seed: Optional[int] = None) -> None:
    if capacity < 0:
      raise ValueError("capacity must be non-negative")
    self.capacity: int = capacity
    self.seen: int = 0
    self._reservoir: list[Item] = []
    self._random: random.Random = random.Random(seed)

  def offer(self, item: Item) -> None:
    """
      Present one stream item to the sampler.\n
      While the reservoir is unfilled the item is admitted outright;\n
      afterward it is admitted with probability capacity / seen, evicting\n
      a uniformly random occupant.\n
    """
    # while the reservoir has room, admit the item outright.
    self.seen += 1
    if len(self._reservoir) < self.capacity:
      self._reservoir.append(item)
      return

    # the seen-th item earns a slot with probability capacity / seen.
    position: int = self._random.randint(1, self.seen)
    if position <= self.capacity:
      self._reservoir[position - 1] = item

  def sample(self) -> list[Item]:
    """
      A copy of the current reservoir contents.\n
    """
    return list(self._reservoir)

  def __len__(self) -> int:
    return len(self._reservoir)

def reservoir_sample(
  stream: Iterable[Item],
  capacity: int,
  seed: Optional[int] = None,
) -> list[Item]:
  """
    Draw a uniform random `capacity`-subset of `stream` in one pass.\n
    Convenience wrapper over ReservoirSampler for an entire iterable.\n
  """
  sampler: ReservoirSampler[Item] = ReservoirSampler(capacity, seed)
  for item in stream:
    sampler.offer(item)
  return sampler.sample()

Morris counting: a count in $O(\log \log N)$ bits

Problem. Count events up to $N$, but spend far fewer than the ${\log }_{2}N$ bits an exact counter needs.

Idea. Don't store the count $c$; store an exponent $X$ that approximates ${\log }_{2}c$. On each increment, advance $X$ only probabilistically, with probability $2^{-X}$, so that $X$ grows by one roughly every time the true count doubles. The estimate read out is $\hat{c}=2^X-1$.

Because $X$ only needs to reach about ${\log }_{2}N$, storing $X$ takes $O(\log \log N)$ bits, an exponential saving over the counter it approximates. Counting to a billion takes $X\le 30$, which fits in $5$ bits.

A worked trace. One possible run of ten increments:

After ten increments the estimate reads $7$ against a true count of $10$: off by $30\%$, which is typical, since a single Morris counter has constant relative error. The estimate is coarse but never drifts systematically, and that is the precise content of the next claim.

Variance, and how averaging fixes it. The same conditioning computes the second moment. With $Z_c=4^{X_c}$,

so $\mathbb{E}[Z_{c+1}]=\mathbb{E}[Z_c]+3\mathbb{E}[Y_c]=\mathbb{E}[Z_c]+3(c+1)$, and summing from $Z_0=1$ gives $\mathbb{E}[4^{X_c}]=1+\frac{3c(c+1)}{2}$. Therefore

The standard deviation is about $c/\sqrt{2}$, comparable to the count itself, so one counter is only good to a constant factor. Chebyshev's inequality makes the repair quantitative: averaging $m$ independent counters divides the variance by $m$, so

which drops below $\delta$ once $m\ge 1/(2{\epsilon }^2\delta )$. The total space is $O({\epsilon }^{-2}{\delta }^{-1}\log \log N)$ bits, still exponentially below an exact counter for fixed accuracy.² Alternatively, replacing base $2$ by $1+1/b$ makes $X$ advance more often and shrinks the per-counter variance at the cost of a slightly larger register.

Morris counting anticipates the sketches that follow: a single small random variable standing in for a quantity too large to store exactly, made accurate by averaging independent copies.³

import random
from typing import Optional

class MorrisCounter:
  """
    A single probabilistic counter storing only an integer exponent.\n
    `increment` advances the exponent with probability 2^-exponent;\n
    `estimate` reads out 2^exponent - 1.\n
  """

  def __init__(self, seed: Optional[int] = None) -> None:
    self.exponent: int = 0
    self._random: random.Random = random.Random(seed)

  def increment(self) -> None:
    """
      Register one event, raising the exponent with probability\n
      2^-exponent so it advances about once per doubling of the count.\n
    """
    if self._random.random() < 2.0 ** (-self.exponent):
      self.exponent += 1

  def estimate(self) -> float:
    """
      The unbiased count estimate 2^exponent - 1.\n
    """
    return 2.0**self.exponent - 1.0

class AveragedMorrisCounter:
  """
    A bank of independent Morris counters whose estimates are averaged.\n
    Averaging `trials` copies shrinks the variance by a factor `trials`,\n
    trading a little more space for a tighter relative error.\n
  """

  def __init__(self, trials: int = 16, seed: Optional[int] = None) -> None:
    if trials < 1:
      raise ValueError("trials must be at least 1")
    base: random.Random = random.Random(seed)
    self._counters: list[MorrisCounter] = [
      MorrisCounter(base.randrange(2**31)) for _ in range(trials)
    ]

  def increment(self) -> None:
    """
      Register one event in every underlying counter.\n
    """
    for counter in self._counters:
      counter.increment()

  def estimate(self) -> float:
    """
      The mean of the per-counter estimates.\n
    """
    total: float = sum(counter.estimate() for counter in self._counters)
    return total / len(self._counters)

Where sampling and counting run

Reservoir sampling and Morris counting are old ideas, 1985 and 1978 respectively, that never went away, because the problems they solve, sampling a stream whose length you learn only at the end and counting without room for a counter, keep recurring at scale.

Reservoir sampling (Vitter, 1985, who named it and gave the skip-ahead optimization) is the standard way to draw a fair sample from a log you cannot buffer: distributed systems use it to keep a bounded, uniform sample of requests for tracing and A/B analysis, and Apache Kafka and Spark ship variants. The weighted generalization, keying each item by $u_i^{1/w_i}$ and keeping the top keys, is the A-Res algorithm of Efraimidis and Spirakis (2006), used when items should be sampled with unequal probability, such as sampling clicks in proportion to dwell time.

Morris counting (Morris, 1978, analyzed rigorously by Flajolet in 1985) foreshadowed a whole family of probabilistic counters. Its modern descendants are the approximate counters inside monitoring systems that must track billions of events per second in a fixed register budget, where the constant relative error is a fair price for shrinking a $\log N$-bit counter to $\log \log N$ bits. The averaging trick that repairs its variance, run $m$ independent copies and average, is the same move that turns the noisy sketches of the next lesson into trustworthy estimators.⁴

Continuing on to sketches

Reservoir sampling and Morris counting each summarized the stream as a whole, a representative subset, a single approximate total. The harder and more common questions are about the stream's contents: how often did a particular item appear, which items are frequent, how many distinct items were there? Answering those in sublinear space calls for sketches, small hashed counter arrays with the same exactness-for-space trade-off pushed one level further. This continues in Streaming Sketches.

Takeaways

• The streaming model processes $x_1,\dots ,x_n$ in one pass, keeping only $\mathrm{polylog}(n)$ state; items are seen once and discarded. An information-theoretic argument forces exact answers to cost $\Omega (n)$ space, so the model trades exactness for approximate, probabilistic answers.
• The three cost models line up by what they assume about the data: RAM (fits, freely re-readable, spends time, exact), external memory (re-readable but too big, spends block transfers, exact), streaming (seen once, spends space, approximate).
• Reservoir sampling keeps a uniform $k$-sample of an unknown-length stream in $O(k)$ space by admitting item $i$ with probability $k/i$; the telescoping product $\frac{k}{i}{\prod }_{t>i}\frac{t-1}{t}=\frac{k}{n}$ makes every item equally likely to be retained, at every prefix. The case $k=1$ is a uniform pick from a list of unknown length in $O(1)$ space.
• Morris counting approximates a count up to $N$ in $O(\log \log N)$ bits with the unbiased estimator $2^X-1$; its variance $c(c-1)/2$ is cut by averaging independent copies, the same technique the sketches of the next lesson use.
• These two summarize the stream as a whole; estimating its per-item frequencies needs the sketches that follow.