Streaming Statistics Toolkit

Define the Toolkit and Its Users

A streaming-statistics toolkit is like a dashboard at a busy railway station: it cannot store every passenger forever, but it must continuously answer useful questions—how many arrived, how many were distinct, whether a name has probably appeared, which destinations are busiest, and where the wait-time distribution lies. Producers send events, applications ask questions, and operators need to know whether an answer is exact, approximate, stale, or degraded.

Start with the stream A, B, A, C, A. An exact counter reports 5 events; a distinct set reports 3 keys; a frequency map reports A → 3; a membership structure answers that B was seen. An approximate toolkit may replace those structures with a Morris counter, HyperLogLog (HLL), Count-Min Sketch, and Bloom filter to use bounded memory. The output A ≈ 3 is only useful if it also states the configured error and confidence.

The common contract is update(event), query(request), merge(other), snapshot(), and restore(bytes). A parameter calculator turns an error budget into memory settings. An exact oracle processes the same replay with dictionaries, sets, or sorted arrays. Confidence metadata describes a probabilistic statement; it is not a promise that every answer is within the bound. Mergeability requires identical algorithm families, seed identities, schema versions, and compatible parameters—not merely two objects with the same method names.

The users are application engineers choosing a statistic, data-platform engineers deploying partitioned state, reliability engineers diagnosing lag or corruption, security reviewers limiting sensitive output, and analysts interpreting estimates. Acceptance means stable interfaces, deterministic seeded tests, bounded memory, merge-equivalent results where promised, versioned serialization, visible degradation, and exact-oracle evidence for every approximate backend.

Choose Interfaces, Guarantees and Backends

The primary workload is keyed events on many partitions, with millions of updates per second, point queries below 20 ms at p99, snapshots every five minutes, and bounded per-tenant memory. Producers submit {tenant, stream, event_id, event_time, key, value, schema_version}. The service validates size and type, applies an explicit duplicate policy, routes by tenant and stream, and returns a Result containing value, algorithm, exactness, error model, confidence, parameters, watermark, and state version.

Functional requirements include counts, uniform samples, membership, distinct counts, point frequencies, heavy-hitter candidates, and quantiles. SLOs include 99.9% query availability, no silent restore across incompatible versions, deterministic replay under fixed seeds, and resource enforcement before allocation. Privacy limits prohibit raw sampled payloads unless the tenant policy permits them; access control, retention, redaction, and query auditing are part of the contract. The threat model covers malformed events, hash-flooding, cross-tenant reads, snapshot tampering, replay duplication, and a compromised operator credential, but not a fully compromised process host.

Parameter calculators translate targets into configuration: Bloom uses m = ceil(-n ln p/(ln 2)^2) bits and k ≈ (m/n)ln 2; CMS uses width ceil(e/ε) and depth ceil(ln(1/δ)); HLL precision selects m=2^p registers with relative standard error near 1.04/sqrt(m). Values outside safe ranges are rejected. Exact backends remain available for small streams, audits, and automatic fallback when an approximation is unsupported—not when memory would become unsafe.

flowchart TD
    P[Producers] --> V[Validate and route]
    V --> L[Partition log]
    L --> E[State engine]
    E --> X[Exact backends]
    E --> A[Approximate backends]
    X --> S[Versioned snapshots]
    A --> S
    Q[Query API] --> E
    E --> R[Result plus guarantee metadata]
    O[Operators] --> C[Config and rollout control]
    C --> V
    E --> M[Metrics traces audit]
    classDef navy fill:#17365D,color:#fff,stroke:#102844;
    classDef teal fill:#147D84,color:#fff,stroke:#0c5c61;
    classDef crimson fill:#C6284D,color:#fff,stroke:#8e1935;
    classDef purple fill:#6C3BAA,color:#fff,stroke:#482675;
    class P,Q,O navy;
    class V,L,E,C teal;
    class X,A,S purple;
    class R,M crimson;

Build Exact and Approximate Statistics Vertically

Build one statistic end to end before adding the next. Slice 1 implements the envelope, validation, exact count, snapshot checksum, and replay. Slice 2 adds Bloom membership with a calculator and exact-set oracle. Slice 3 adds reservoir sampling, then HLL, CMS/Count Sketch, and a quantile summary. Each backend conforms to a narrow protocol but may expose statistic-specific queries; forcing membership and quantiles into the same scalar method would hide semantics.

Configuration is immutable within a state generation. A snapshot records {format_version, algorithm, parameters, seed_id, schema_version, watermark, payload, checksum}. Restore first validates size, checksum, version, and parameter compatibility. Migrators are explicit functions between adjacent formats; otherwise the service replays the retained log or returns UNSUPPORTED_STATE. Memory guards check calculated dimensions before allocation. Counter saturation, late-event policy, and serialization limits are observable. A circuit breaker can route a small stream to an exact backend, reject optional statistics, or serve a stale labeled snapshot; it never silently changes guarantees.

from dataclasses import dataclass
from hashlib import blake2b
from random import Random
from typing import Any

def _hash(seed: int, value: str) -> int:
    return int.from_bytes(blake2b(f"{seed}:{value}".encode(), digest_size=8).digest(), "big")

@dataclass(frozen=True)
class Config:
    width: int = 128
    depth: int = 5
    reservoir_size: int = 8
    seed: int = 7
    max_cells: int = 1_000_000
    format_version: int = 1

    def validate(self) -> None:
        if not (1 <= self.width and 1 <= self.depth and self.width * self.depth <= self.max_cells):
            raise ValueError("unsafe Count-Min dimensions")
        if not 1 <= self.reservoir_size <= 100_000:
            raise ValueError("unsafe reservoir size")
        if self.format_version != 1:
            raise ValueError("unsupported format version")

class Toolkit:
    def __init__(self, config: Config):
        config.validate()
        self.config = config
        self.count = 0
        self.exact: dict[str, int] = {}
        self.cms = [[0] * config.width for _ in range(config.depth)]
        self.sample: list[Any] = []
        self.rng = Random(config.seed)

    def update(self, key: str, payload: Any = None) -> None:
        if not isinstance(key, str) or not key or len(key.encode()) > 256:
            raise ValueError("key must be 1..256 UTF-8 bytes")
        self.count += 1
        self.exact[key] = self.exact.get(key, 0) + 1
        for row in range(self.config.depth):
            col = _hash(self.config.seed + row, key) % self.config.width
            self.cms[row][col] += 1
        item = payload if payload is not None else key
        if len(self.sample) < self.config.reservoir_size:
            self.sample.append(item)
        else:
            slot = self.rng.randrange(self.count)
            if slot < self.config.reservoir_size:
                self.sample[slot] = item

    def frequency(self, key: str) -> dict[str, Any]:
        estimate = min(self.cms[row][_hash(self.config.seed + row, key) % self.config.width]
                       for row in range(self.config.depth))
        return {"value": estimate, "algorithm": "count_min", "exact": False,
                "epsilon_upper": 2.718281828 / self.config.width,
                "delta": 2.718281828 ** (-self.config.depth),
                "format_version": self.config.format_version}

    def merge(self, other: "Toolkit") -> "Toolkit":
        if self.config != other.config:
            raise ValueError("incompatible state: config or seed differs")
        merged = Toolkit(self.config)
        merged.count = self.count + other.count
        merged.exact = self.exact | other.exact
        for key in set(self.exact) | set(other.exact):
            merged.exact[key] = self.exact.get(key, 0) + other.exact.get(key, 0)
        merged.cms = [[a + b for a, b in zip(left, right)]
                      for left, right in zip(self.cms, other.cms)]
        # A production merge uses a priority reservoir; refusing here is safer
        # than claiming concatenation is uniform.
        merged.sample = []
        return merged

def self_test() -> None:
    cfg = Config(width=64, depth=4, reservoir_size=3, seed=11)
    a, b = Toolkit(cfg), Toolkit(cfg)
    for key in ["A", "B", "A"]: a.update(key)
    for key in ["C", "A"]: b.update(key)
    whole = Toolkit(cfg)
    for key in ["A", "B", "A", "C", "A"]: whole.update(key)
    merged = a.merge(b)
    assert whole.count == merged.count == 5
    assert whole.exact == merged.exact == {"A": 3, "B": 1, "C": 1}
    assert whole.cms == merged.cms
    assert whole.frequency("A")["value"] >= 3
    assert Toolkit(cfg).sample == []
    try:
        Toolkit(Config(width=10**6, depth=2))
    except ValueError:
        pass
    else:
        raise AssertionError("resource guard failed")
    try:
        a.merge(Toolkit(Config(width=64, depth=4, reservoir_size=3, seed=12)))
    except ValueError:
        pass
    else:
        raise AssertionError("compatibility guard failed")
    print("all integration checks passed", whole.frequency("A"))

if __name__ == "__main__":
    self_test()

flowchart TD
    C[Contract and envelope] --> E[Exact baseline]
    E --> B[Bloom and reservoir]
    B --> H[HLL and frequency sketches]
    H --> Q[Quantile summary]
    Q --> I[Integrated canary]
    I --> D[Progressive deployment]
    D -->|SLOs pass| P[Promote]
    D -->|Mismatch or saturation| R[Rollback generation]
    R --> E
    classDef navy fill:#17365D,color:#fff,stroke:#102844;
    classDef teal fill:#147D84,color:#fff,stroke:#0c5c61;
    classDef crimson fill:#C6284D,color:#fff,stroke:#8e1935;
    classDef purple fill:#6C3BAA,color:#fff,stroke:#482675;
    class C,E navy;
    class B,H,Q teal;
    class I,D purple;
    class P,R crimson;

Verify Every Contract and Failure Mode

Verification layers reinforce one another. Unit tests cover parameter formulas and serialization. Property tests generate streams and assert exact identities, Bloom no-false-negative behavior, CMS non-underestimation for nonnegative updates, merge equivalence, and bounded state. Differential replay compares approximate answers with exact oracles across seeds and skew levels. Load tests measure update throughput, query p99, memory per stream, snapshot duration, and recovery time. Adversarial tests target hot keys, empty streams, Unicode, maximum counters, late events, duplicate IDs, incompatible seeds, corrupt snapshots, storage stalls, and partition loss.

An approachable end-to-end case replays 10,000 café orders and checks count, distinct menu items, frequent products, a uniform audit sample, and p95 preparation time. The professional case is a multi-tenant telemetry service at 500,000 events/s, 5,000 active streams, a 4 GiB state budget per worker, query p99 under 20 ms, and recovery under 10 minutes. Canary traffic shadows exact results for small tenants. Dashboards separate algorithm error from systems error: lag, rejected events, saturation, incompatible merges, checksum failures, and estimate/oracle deltas. An incident commander owns rollback; the correctness lead signs the dossier; security reviews sampled data and membership-inference exposure.

Package, Demonstrate and Extend the Toolkit

The release bundle contains an API reference with semantic examples, configuration calculators, a pinned reproducible environment, seeded datasets, benchmark commands, raw results, a correctness dossier, schema and migration notes, an operational runbook, security/privacy review, limitations, and a roadmap. The runbook covers saturation, lag, snapshot corruption, incompatible state, elevated false positives, unexpected oracle drift, tenant isolation, rollback, and evidence preservation. Every artifact names its version, owner, generation command, and review date.

The demonstration starts with the five-event example, shows exact and approximate result envelopes, partitions and merges a replay, rejects an incompatible seed, restores a valid snapshot, rejects a corrupt one, and finishes with the professional workload dashboard. Peer reviewers should ask: Can a caller distinguish every guarantee? Are calculators tested at boundary values? Is merge equivalence proved or explicitly unsupported? Can operators see degradation before users do? Are samples and membership queries appropriately protected? Could a new backend be added without weakening old contracts?

Handoff requires tagged source, checksums, a clean one-command test run, benchmark provenance, reviewed dashboards and alerts, rehearsed rollback, ownership rotation, and no undocumented exceptions. The roadmap prioritizes risk reduction before feature breadth: priority-reservoir merge correctness, production HLL bias tables, KLL quantiles, differential-privacy wrappers with explicit budgets, hardware-aware vectorization, and cross-language conformance fixtures.