Distributed Streaming Analytics Engine

Write the Distributed Analytics Charter

A distributed streaming analytics engine is like a network of local election offices. Each office processes only its assigned ballots, records exactly which batch it covered, and sends a compatible subtotal up a merge tree. The national result is meaningful only if every expected region appears once, no region appears twice, the rules match, and uncertainty from approximate subtotals is carried forward rather than erased.

Consider events A, B, A, C, A, D split by stable routing into two partitions: P0 receives A,A,A, P1 receives B,C,D. Local counts are 3 and 3, so the merged count is 6; local frequency maps merge to {A:3,B:1,C:1,D:1}; distinct sets union to four keys. An HLL may merge registers instead, a Count-Min Sketch adds aligned cells, a KLL merges compactor state, and reservoirs require merge priorities. “Has a merge method” is therefore not enough—the algebra, parameters, seeds, window, epoch coverage, and failure model must match.

A pluggable sketch implements a declared lifecycle without hiding its specific guarantee. Partitioning assigns each logical update to an owner. A merge tree combines partial states hierarchically to reduce fan-in. A window limits event-time scope. Persistence checkpoints versioned state and replay position. Versioning covers schema, algorithm, parameters, seeds, and policy. Privacy constrains collection, state, queries, and releases. Robustness limits malformed, adversarial, or adaptive inputs. Observability explains lag, state, error, merge coverage, and degradation. Evaluation compares estimates with exact oracles and SLOs.

Stakeholders include product teams, analysts, stream-platform engineers, SREs, privacy/security reviewers, finance, and incident commanders. Acceptance requires deterministic routing and replay, complete non-overlapping merge coverage, bounded per-tenant resources, explicit event/processing-time rules, version-compatible state, calibrated accuracy, governed access, resilient recovery, exact shadow evidence, and result envelopes that distinguish estimate error from incomplete or stale execution.

Design Partitions, Merge Trees and Trust Boundaries

The target platform ingests 2 million events/s across thousands of tenants and supports frequency, cardinality, quantile, sampling, similarity, matrix, graph, and privacy-aware aggregates. Producers submit {tenant, stream, update_id, event_time, routing_key, kind, payload, schema_version}. The gateway authenticates, validates and redacts, applies quotas, assigns a routing epoch, and appends to a durable partition log. Workers maintain windowed plugin states and checkpoint offsets. Regional combiners form a balanced merge tree; a root coordinator validates coverage before publishing a result.

Each plugin declares stream model (cash-register, strict/general turnstile, row, edge), partition key, capabilities, parameters, seeds, merge operation, serialization format, accuracy/confidence metadata, memory estimator, privacy classification, and migration rules. Frequency uses CMS or Count Sketch according to update semantics; cardinality uses HLL; quantiles use KLL/GK; sampling uses priority/reservoir methods; similarity uses signature/LSH plus exact verification; matrix and graph services expose their own low-rank/recovery contracts. The engine standardizes lifecycle and evidence, not mathematical meaning.

SLOs are 99.99% accepted-event durability, update p99 below 10 ms, query p99 below 150 ms, complete-window publication within 15 seconds of watermark closure, checkpoint below 30 seconds, RPO zero for acknowledged log events, recovery below 10 minutes, and hard worker/tenant budgets. Threats include malformed payloads, duplicate/reordered events, hot keys, state explosion, forged tenant scope, hash attacks, adaptive queries, snapshot tampering, Byzantine/stale workers, split-brain ownership, version downgrade, and privacy-budget bypass.

flowchart TD
    P[Producers and SDKs] --> G[Auth validate redact]
    G --> L[Partitioned durable log]
    L --> W[Stateful partition workers]
    W --> S[Versioned sketch services]
    S --> C[Regional combiners]
    C --> R[Root merge and coverage gate]
    R --> Q[Query result envelopes]
    W --> K[Checkpoint store]
    O[Operators] --> T[Config privacy rollout control]
    T --> G
    T --> S
    W --> M[Metrics traces audit]
    R --> M
    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,O navy;
    class G,L,W,T teal;
    class S,C,K purple;
    class R,Q,M crimson;

Integrate Versioned Sketch Services

Build vertical slices. Slice 1 routes a typed event to exact count, set, frequency, and quantile oracles on two partitions, then performs a coverage-gated merge. Slice 2 adds checkpoint offsets, idempotency, event-time windows, watermarking, and recovery. Slice 3 introduces one approximate plugin at a time—HLL, CMS/Count Sketch, KLL, sampling—while shadowing exact small tenants. Slice 4 adds similarity, matrix, graph, and privacy-aware services with their own capability contracts. Slice 5 adds dual-write upgrades, merge-tree canaries, resharding, and safe degradation.

The stable runtime interface is create(config), apply(update), query(request), merge(states, coverage), snapshot(), restore(), and describe_guarantee(). Configuration validation estimates memory and CPU before allocation. A state identity includes tenant, stream, plugin, schema, algorithm, parameters, seed, routing epoch, window, and checkpoint generation. Only adjacent tested migrations are automatic; otherwise the engine replays the retained log. Safe fallback may use a stale complete checkpoint, exact bounded state, partial labelled result for approved consumers, or explicit rejection—never a silent algorithm or privacy change.

from dataclasses import dataclass
from collections import Counter
from hashlib import blake2b
from typing import Iterable

def route(key: str, partitions: int, epoch_seed: int) -> int:
    raw = blake2b(f"{epoch_seed}:{key}".encode(), digest_size=8).digest()
    return int.from_bytes(raw, "big") % partitions

@dataclass(frozen=True)
class Config:
    partitions: int = 2
    epoch: int = 1
    epoch_seed: int = 17
    max_keys: int = 10_000
    format_version: int = 1
    def validate(self) -> None:
        if not 1 <= self.partitions <= 4096: raise ValueError("unsafe partition count")
        if self.epoch < 0 or not 1 <= self.max_keys <= 1_000_000: raise ValueError("unsafe limits")
        if self.format_version != 1: raise ValueError("unsupported format version")

class ExactPartition:
    def __init__(self, config: Config, partition_id: int):
        config.validate()
        if not 0 <= partition_id < config.partitions: raise ValueError("invalid partition")
        self.config, self.partition_id = config, partition_id
        self.count, self.frequencies, self.seen = 0, Counter(), set()

    def apply(self, update_id: str, key: str) -> str:
        if not update_id or not key: raise ValueError("invalid update")
        if route(key, self.config.partitions, self.config.epoch_seed) != self.partition_id:
            raise ValueError("routing ownership violation")
        if update_id in self.seen: return "duplicate"
        if key not in self.frequencies and len(self.frequencies) >= self.config.max_keys:
            raise MemoryError("key budget exceeded")
        self.seen.add(update_id); self.count += 1; self.frequencies[key] += 1
        return "accepted"

@dataclass
class MergedResult:
    count: int
    frequencies: Counter
    coverage: frozenset[int]
    epoch: int
    complete: bool

def merge(states: Iterable[ExactPartition], expected: frozenset[int]) -> MergedResult:
    states = list(states)
    if not states: raise ValueError("no states")
    config = states[0].config
    if any(state.config != config for state in states): raise ValueError("incompatible state")
    coverage = [state.partition_id for state in states]
    if len(coverage) != len(set(coverage)): raise ValueError("duplicate partition coverage")
    actual = frozenset(coverage)
    if not actual <= expected: raise ValueError("unexpected partition")
    frequencies = Counter()
    for state in states: frequencies.update(state.frequencies)
    return MergedResult(sum(s.count for s in states), frequencies, actual,
                        config.epoch, actual == expected)

class Engine:
    def __init__(self, config: Config):
        config.validate(); self.config = config
        self.states = [ExactPartition(config, i) for i in range(config.partitions)]
    def apply(self, update_id: str, key: str) -> str:
        return self.states[route(key, self.config.partitions, self.config.epoch_seed)].apply(update_id, key)
    def result(self) -> MergedResult:
        return merge(self.states, frozenset(range(self.config.partitions)))

def self_test() -> None:
    cfg = Config(partitions=2, epoch=7, epoch_seed=23)
    engine = Engine(cfg)
    for i, key in enumerate(["A","B","A","C","A","D"]):
        assert engine.apply(str(i), key) == "accepted"
    assert engine.apply("0", "A") == "duplicate"
    result = engine.result()
    assert result.count == 6 and result.frequencies == Counter(A=3,B=1,C=1,D=1)
    assert result.complete and result.coverage == frozenset({0,1})
    partial = merge([engine.states[0]], frozenset({0,1}))
    assert not partial.complete and partial.coverage == frozenset({0})
    try: merge([engine.states[0], engine.states[0]], frozenset({0,1}))
    except ValueError: pass
    else: raise AssertionError("duplicate coverage accepted")
    incompatible = ExactPartition(Config(partitions=2, epoch=8, epoch_seed=23), 0)
    try: merge([engine.states[0], incompatible], frozenset({0,1}))
    except ValueError: pass
    else: raise AssertionError("cross-epoch merge accepted")
    print("integration checks passed", result.count, dict(result.frequencies), sorted(result.coverage))

if __name__ == "__main__": self_test()

flowchart TD
    C[Typed events and exact oracles] --> P[Two-partition coverage merge]
    P --> W[Windows watermarks checkpoints]
    W --> B[Core sketch plugins]
    B --> A[Advanced similarity matrix graph]
    A --> D[Privacy and robustness policies]
    D --> U[Dual-write upgrade]
    U --> K[Shadow merge-tree canary]
    K -->|Coverage oracle SLO pass| R[Progressive promotion]
    K -->|Mismatch or drift| X[Rollback generation]
    X --> P
    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,P navy;
    class W,B,A,D teal;
    class U,K purple;
    class R,X crimson;

Prove Correctness Under Replay, Failure and Upgrade

Verification composes local algorithm properties with distributed invariants. Unit/property tests cover routing determinism, exclusive ownership, idempotency, window boundaries, watermark monotonicity, merge associativity where promised, coverage union/disjointness, serialization, migration, privacy accounting, and result envelopes. Differential replay compares plugin outputs with exact maps, sets, sorted arrays, all-pairs similarity, Gram/SVD calculations, and BFS/graph oracles on shadow-sized workloads.

The approachable case is the six-event, two-partition trace. The professional workload drives 2 million events/s across 256 partitions, 10,000 tenants, mixed windows, skewed keys, late/reordered updates, merge fan-in, checkpointing, resharding, and all plugin families. SLOs include update p99 <10 ms, query p99 <150 ms, 99.99% accepted-event durability, complete results ≤15 seconds after window closure, worker memory within configured budget, zero unlabelled partial results, checkpoint <30 seconds, recovery <10 minutes, and accuracy/privacy guarantees inside declared envelopes.

Failure injection kills workers and combiners, partitions regions, duplicates log segments, stalls checkpoint storage, changes router seeds, corrupts state bytes, creates hot tenants, exhausts privacy budgets, adapts queries against fixed sketches, and interrupts dual-write upgrades. Metrics expose lag, watermark delay, dedup hits, ownership violations, state bytes, merge depth/time, missing/duplicate coverage, version mismatch, plugin saturation/error, privacy spend, checkpoint age, shadow divergence, and degraded responses. Traces carry update ID, partition, epoch, window, plugin, and merge node; audit logs preserve access, config, budgets, upgrade decisions, incidents, and acknowledgements.

Deliver the Engine and Its Evolution Roadmap

The release bundle contains event/query/result schemas, routing and merge specifications, plugin capability contracts, parameter calculators, pinned environments, seeded datasets, benchmark commands/raw results, correctness and privacy dossiers, migration matrix, topology diagrams, dashboards, alerts, and the operational runbook. Every artifact names its version, owner, generation command, review date, and evidence links. The public API documents exact, approximate, incomplete, stale, degraded, and rejected outcomes separately.

The runbook begins with lag, hot partitions, missing/duplicate coverage, plugin saturation, privacy-budget exhaustion, state incompatibility, checkpoint corruption, split brain, query drift, and dual-write divergence. It gives diagnostics, thresholds, safe fallback, escalation, fencing, rollback, and evidence-preservation procedures. The demonstration routes and merges the six-event example, withholds a partition to show partial coverage, replays duplicates, rejects a cross-epoch merge, restores a checkpoint, runs one approximate plugin beside an oracle, performs a dual-write upgrade, and finishes on the professional SLO/cost dashboard.

Peer reviewers ask: Does each plugin declare its actual merge algebra? Can every global result prove coverage? What happens during resharding and late data? Are algorithm error and distributed incompleteness distinguishable? Can privacy releases be duplicated through retries? Are advanced similarity/matrix/graph workflows forced into invalid common semantics? Handoff requires tagged code/images, checksums, clean one-command replay, benchmark provenance, signed dossiers, rehearsed incident/upgrade rollback, on-call ownership, and no undocumented bypass.