Streaming and real-time ML

Event time is when an event actually occurred at the source, embedded as a timestamp in the event payload, while processing time is when the stream engine receives and processes the event. Event-time processing produces deterministic, reproducible results regardless of network delays or reprocessing, but requires watermarks to handle late-arriving data. Processing-time windows are simpler to implement but non-deterministic and will produce different results on replay.

A watermark in Flink is a monotonically increasing timestamp injected into the stream that declares 'no events with a timestamp earlier than W will arrive.' It allows the engine to close event-time windows and emit results without waiting indefinitely for late data. Watermarks trade completeness for latency — a larger watermark delay tolerates more late events but increases output latency proportionally.

A sliding window (also called a hopping window) with a size of 5 minutes and a hop/slide interval of 30 seconds. Tumbling windows do not overlap and emit results only at window boundaries every 5 minutes, providing no intermediate updates. Session windows are event-gap-driven and unsuitable for fixed-interval aggregations.

At-least-once semantics guarantee every message is delivered but allow duplicates when producers retry after failures. Exactly-once semantics (EOS) use idempotent producers (sequence numbers per partition) combined with the Kafka Transactions API to ensure each message is committed exactly once end-to-end — at the cost of measurably higher latency and reduced throughput due to transactional coordination overhead (typically 20–100% overhead depending on configuration and batch size). EOS is essential for financial aggregations or counters, but overkill for approximate analytics where duplicates are tolerable.

Each partition in a Kafka topic can be assigned to exactly one consumer within a group, so adding consumers (up to the number of partitions) increases parallel processing throughput linearly. The group coordinator manages partition assignment and triggers rebalancing when consumers join or leave. This design decouples producers from consumers and allows independent scaling of each without changing producer code.

ADWIN (ADaptive WINdowing) maintains a variable-length sliding window over the stream and detects drift by testing whether the means of two sub-windows are statistically different beyond a Hoeffding-bound threshold; when drift is detected, it shrinks the window to discard stale data. Page-Hinkley is a lightweight cumulative-sum test best suited for detecting gradual monotonic shifts in a single statistic with a fixed sensitivity parameter. ADWIN is preferred when the drift magnitude is unknown, when drift speed varies over time, or when the algorithm must self-adapt its window size rather than requiring manual threshold tuning.

Calling predict_one first generates a prediction using the current model state before the true label is revealed, which is the correct online evaluation protocol. Calling learn_one first updates the model weights with the true label before predicting, which allows the model to 'see the answer' before evaluation — a form of data leakage that inflates reported accuracy and does not reflect real deployment behavior. The test-then-train (prequential) order is the standard in online learning.

from river import linear_model, metrics

model = linear_model.LogisticRegression()
metric = metrics.Accuracy()

for x, y in stream:
    y_pred = model.predict_one(x)   # predict FIRST
    metric.update(y, y_pred)        # evaluate before label is used
    model.learn_one(x, y)           # then update the model

Lambda architecture combines a batch layer that reprocesses full historical data for accuracy with a speed layer that processes the real-time stream for low latency, merging both outputs in a serving layer. It solves the tension between the accuracy of full-batch recomputation and the low latency of streaming. The major criticism is that maintaining two separate codebases for the same business logic — one batch (MapReduce/Spark), one streaming (Flink/Storm) — doubles operational complexity and introduces subtle inconsistencies; the Kappa architecture addresses this by replacing both layers with a single streaming pipeline backed by a replayable log (Kafka), using stream replay for historical reprocessing.

Backpressure occurs when a downstream operator cannot keep up with upstream throughput, causing upstream buffers to fill. Flink handles backpressure via a credit-based flow control mechanism built into its network stack: upstream operators only transmit data when downstream operators signal that they have available buffer credit. This propagates backpressure naturally from the slowest operator all the way to the source without dropping records, causing OOM errors, or requiring explicit throttling logic in user code.

Spark Streaming (DStreams) treated a stream as a sequence of RDD micro-batches at a fixed interval, exposing low-level RDD APIs with separate batch and streaming codebases. Structured Streaming models the stream as an unbounded DataFrame/Dataset, using the same SQL/DataFrame API as batch Spark and internally executing micro-batches (or continuous processing in experimental mode) with end-to-end exactly-once guarantees via offset tracking and idempotent sinks. This unified model eliminates dual API maintenance and allows the same transformation code to run on batch and streaming data without modification.

A feature store is a centralized platform (e.g., Feast, Tecton, Hopsworks) that stores, versions, and serves ML features for both offline training and online inference. A point-in-time correct join ensures that when constructing a training example labeled at time T, only feature values that were available before T are joined — without it, future feature values bleed into training labels, causing training-serving skew and overly optimistic offline metrics that do not generalize to production. This leakage is especially severe in streaming scenarios where feature values change at sub-second granularity.

The Very Fast Decision Tree (VFDT) uses the Hoeffding bound — a statistical guarantee that, with high probability, the attribute chosen as the best split by a small sample will also be the best choice given the full population. This allows the tree to make split decisions using only sufficient statistics (counts, sums) accumulated per leaf rather than storing all past examples. Each example is processed in O(tree depth) time — O(log N) on average for a balanced tree — memory usage is bounded by the number of active leaves regardless of stream length, and the model updates incrementally without any retraining from scratch, making it the reference baseline for streaming classification tasks.

CDC captures row-level changes (inserts, updates, deletes) in a source database by reading its transaction log (e.g., PostgreSQL WAL, MySQL binlog) via tools like Debezium, without polling or impacting the source DB under load. These change events are published to Kafka, enabling downstream feature computation engines (e.g., Flink, Kafka Streams) to react to database state changes with sub-second latency and update feature store values in real time. This pattern is standard in fraud detection and personalization systems where stale features can degrade model decisions.

Use a stateful operator keyed by user ID that maintains a time-ordered deque (or sorted list) of event timestamps in state. For each new event, append its timestamp, evict entries older than 10 minutes, and emit the deque length as the feature value. In Apache Flink, this is implemented as a KeyedProcessFunction with ListState and registered event-time timers to trigger cleanup of expired entries. This approach uses O(k) memory where k is the maximum events per user per window, avoiding any re-scan of history.

class TxnCounter(KeyedProcessFunction):
    def open(self, ctx):
        desc = ListStateDescriptor('ts', Types.LONG())
        self.state = self.get_runtime_context().get_list_state(desc)

    def process_element(self, value, ctx):
        now = ctx.timestamp()
        window_ms = 10 * 60 * 1000
        valid = [t for t in self.state.get() if t > now - window_ms]
        valid.append(now)
        self.state.update(valid)
        yield (value['user_id'], len(valid))

The learning rate controls the step size of each weight update — too large causes oscillation and instability, too small causes slow adaptation to concept drift. Adagrad and its descendants (Adam, RMSProp) are the preferred adaptive optimizers for sparse, high-cardinality features in streaming settings because they scale per-parameter learning rates inversely with accumulated gradient magnitudes, giving large effective learning rates for rare features and smaller rates for frequent ones. River implements these as `optim.Adagrad`, `optim.Adam`, and `optim.RMSProp`.

Flink aligns sink output with checkpoint barriers: during the pre-commit phase, the sink begins a transaction in the external system and writes data to it without committing; once all operators confirm their state has been snapshotted (checkpoint complete), Flink calls commit() to finalize the transaction. If a failure occurs between pre-commit and commit, on recovery the transaction is aborted in the external system and data is replayed from the last completed checkpoint. This requires the external system to natively support transactions, which Kafka satisfies via its transactional producer API.

A schema registry stores and enforces Avro/Protobuf/JSON Schema definitions for Kafka topics, ensuring that producers and consumers share a contract on data structure without embedding full schemas in every message. In ML pipelines, it prevents silent data corruption when upstream schema changes (added, renamed, or removed fields) break downstream feature extraction or model inference logic. Compatibility modes (backward, forward, full) allow controlled schema evolution, enabling zero-downtime upgrades of producers without taking model-serving consumers offline.

ONNX Runtime (ORT) is a cross-platform inference engine that executes models exported to the Open Neural Network Exchange format, which PyTorch, TensorFlow, and scikit-learn models can all emit. ORT applies graph-level optimizations including operator fusion, constant folding, and memory layout transformations, and supports hardware-specific execution providers (CUDA, TensorRT, CoreML, DirectML) to minimize inference latency. In production real-time systems, serving via ORT typically achieves significantly lower latency than the native training framework because training-time overhead is removed and the inference graph is pre-compiled and optimized.

A stream-table join enriches each streaming event with the latest value from a slowly-changing table (e.g., user profile), is non-windowed, and always reflects the current table state — it is typically backed by a local RocksDB state store with low lookup latency. A stream-stream join matches events from two unbounded streams that arrive within a specified time window, requiring both streams' data to be buffered in state until matches are found or the window expires. Stream-table joins are generally more efficient because only the table side needs full buffering; stream-stream joins buffer both sides.

In offline evaluation on a static test set, the score distribution and population mix are fixed, so PR curves give a controlled comparison. In production, the population shifts over time (concept drift, seasonal patterns), the model threshold may need recalibration on live score distributions, and true labels (confirmed fraud or not) arrive hours or days after prediction — making standard batch PR evaluation misleading. Real-time evaluation must use prequential (test-then-train) windows or time-stratified holdout sets to respect temporal ordering, and must monitor both precision and recall over rolling time windows to detect drift before aggregate AUC metrics degrade.

Micro-batch processing accumulates events over a small interval (e.g., 100ms–1s) and processes them together, achieving higher throughput via vectorized computation and amortized scheduling overhead at the cost of added latency equal to the batch interval. True streaming processes each record individually as it arrives, achieving sub-millisecond latency but lower per-core throughput and higher scheduling overhead per record. Choose micro-batch when throughput and cost efficiency are paramount and 100–500ms latency is acceptable; choose true streaming for interactive fraud checks, real-time recommendation APIs, or gaming applications where p99 latency must be under 10ms.

Prequential evaluation tests the model on each arriving example before using it for a weight update, so every example serves both as an evaluation point and a training point in temporal order — this gives an unbiased performance estimate on non-stationary streams without needing a separate held-out set. Holdout evaluation periodically designates a recent batch of examples as a test set, which wastes labeled data and can be misleading if the holdout distribution diverges from the stream at evaluation time. Prequential is the standard in online learning literature because it respects temporal ordering and uses data maximally efficiently; holdout is sometimes preferred when label noise makes immediate evaluation unreliable.

The heap-based state backend stores all keyed state as Java objects in the JVM heap, providing the lowest read/write latency but bounded by available RAM and subject to garbage collection pauses that can delay checkpoint barriers. RocksDB stores state on local disk (typically SSD) with a configurable block cache in memory, enabling state sizes that far exceed available RAM at the cost of serialization overhead and significantly higher per-read latency for small records. Use heap state for low-latency applications with small state under a few GB; use RocksDB for large user-state tables, long time-window aggregations, or any job where state size exceeds available JVM heap.

CUSUM tracks the cumulative deviation of a monitored statistic (e.g., per-sample model error) from a reference mean, raising an alarm when the cumulative sum exceeds a decision threshold. Unlike simple threshold alerts on individual points, CUSUM aggregates evidence over time, making it sensitive to small but persistent shifts that would not trigger point-in-time alarms. In streaming ML, CUSUM is applied to the rolling sequence of prediction errors; a sustained upward trend in cumulative error signals distribution shift and triggers model retraining or replacement. The key parameter is the allowable slack k (magnitude of shift to tolerate before alarm), which must be tuned to the acceptable drift rate.

A tombstone is a Kafka message with a non-null key and a null value. In log-compacted topics (where Kafka retains only the latest value per key), a tombstone instructs the broker to eventually delete all records for that key during compaction. For stateful streaming ML applications built on log-compacted Kafka topics (e.g., per-user feature stores), producing a tombstone for a user's key implements GDPR 'right to be forgotten' by removing the user's data from the compacted log and triggering cleanup in downstream KTable/state store materializations. Without tombstone support, deleted entities would persist indefinitely in compacted topics.

Pre-compute user and item embeddings offline (or incrementally update via streaming SGD on interaction events) and store them in a low-latency vector store (e.g., Redis with HNSW index, Milvus, or Qdrant). At serve time: retrieve the user embedding via a sub-millisecond key-value lookup, execute ANN search over the item embedding index, apply business filters and ranking, and return results — all within the latency budget. Critical design choices include: keeping the inference path co-located with the vector index to avoid serialization overhead; using HNSW or IVF-PQ indexes for sub-millisecond ANN; sharding item embeddings across nodes to fit within per-node RAM at 100k RPS; and pre-caching frequent user embeddings in L1 memory.

Kafka Streams materializes aggregation state as RocksDB-backed local state stores co-located with the stream processing task on the same JVM, eliminating network round-trips for state reads/writes and enabling microsecond-latency state access. A remote shared database adds network latency (1–20ms) on every state read/write and creates a shared bottleneck that limits horizontal scalability. The trade-off is that local state is partitioned by key (aligned with Kafka partition assignment), so cross-partition queries require either Kafka Streams interactive queries (distributed RPC to the owning node) or a global KTable, whereas a remote database provides global secondary indexing at higher latency cost.

Cold start occurs when a new user, item, or entity has no historical data, making it impossible to produce accurate personalized predictions from a model trained on past behavior. Effective mitigations include: (1) population-level global priors as default predictions, (2) content-based features using only attributes available at entity creation time (demographics, item metadata, text embeddings), (3) bandit algorithms (Thompson Sampling, UCB) applied in a contextual setting to efficiently explore new entities in the live stream with minimum regret, and (4) few-shot or meta-learning approaches pretrained to adapt rapidly from a small number of interactions. The key reframe is to treat cold start as an exploration-exploitation problem rather than a prediction problem.

Flink offers: (1) Allowed lateness — extends a window's lifetime beyond the watermark by a configurable duration so that late elements still trigger window re-evaluation and emit updated results, useful when some lateness is acceptable and downstream consumers can handle result updates; (2) Side outputs — routes elements arriving after the allowed lateness period into a separate tagged stream for independent processing, alerting, or offline reconciliation, suited for financial systems requiring full audit trails; (3) Discard — silently drops late events, acceptable only when the late data rate is negligible and approximate results are sufficient. The right choice depends on accuracy SLA and whether downstream consumers can handle retractions.

The dual write problem occurs when an application must atomically update both a primary database and a message broker (e.g., Kafka) — without a distributed transaction spanning both systems, a crash between the two writes leaves them permanently inconsistent. The outbox pattern resolves this by writing state changes and the outbound event to an 'outbox' table within the same local database transaction; a separate CDC process (e.g., Debezium reading the WAL) then reliably publishes outbox entries to Kafka, providing at-least-once delivery with no inconsistency risk. This is the foundational pattern for building reliable real-time feature pipelines that must stay synchronized between a source-of-truth database and a streaming feature store.

ANN search finds the k most similar vectors to a query without exhaustively comparing all candidates, using index structures like HNSW (graph-based) or IVF-PQ (clustering + quantization). The core trade-off is recall versus latency and scalability: exact search guarantees 100% recall but scales as O(n) per query, making it infeasible for millions or billions of vectors at real-time latency budgets. ANN achieves sub-millisecond query times at billion-vector scale with typical recalls of 95–99%, and the small recall loss is almost always acceptable for recommendation and semantic retrieval tasks where a slightly imperfect result set has negligible business impact.

Data at rest (Kafka topic logs on disk, checkpoint snapshots, training datasets in object storage) should be encrypted with AES-256 — either via Kafka broker-level disk encryption or filesystem/storage-layer encryption. Data in flight (producer-to-broker, broker-to-broker, broker-to-consumer) should use TLS 1.2+ to prevent interception and man-in-the-middle attacks. For PII features (names, emails, behavioral identifiers), field-level encryption before writing to Kafka topics provides defense-in-depth, ensuring that even a broker compromise or misconfigured consumer cannot read sensitive values in plaintext.

Reservoir sampling (Vitter's Algorithm R) fills a buffer of k elements with the first k stream items, then for each subsequent item at position i (i > k), includes it in the reservoir with probability k/i by replacing a uniformly random existing reservoir element. By induction, this guarantees that at any point in the stream every element seen has an equal probability k/n (where n is total elements seen) of being in the reservoir, regardless of stream length or element order. In streaming ML, reservoir sampling is used for maintaining representative training buffers, experience replay in online reinforcement learning, and unbiased population statistics without storing the full stream.

import random

def reservoir_sample(stream, k):
    reservoir = []
    for i, item in enumerate(stream):
        if i < k:
            reservoir.append(item)
        else:
            j = random.randint(0, i)
            if j < k:
                reservoir[j] = item  # replace with prob k/(i+1)
    return reservoir

Model staleness is the performance degradation that occurs as the live data distribution drifts away from the training distribution, with severity proportional to how fast the domain evolves (e.g., financial markets within hours, consumer behavior over days). Four patterns reduce its impact without full retraining: (1) feature freshness — serving real-time features to a stale model often recovers the majority of accuracy loss caused by drift in raw data; (2) incremental fine-tuning — updating only the embedding or output layers on a recent mini-batch rather than the full model; (3) champion-challenger shadow deployment — continuously evaluating a challenger trained on recent data alongside the champion and auto-promoting it when challenger metrics exceed a threshold; (4) bandit-based model routing — distributing traffic proportionally to each model's empirical real-time performance estimate, degrading the stale model's traffic share automatically.