Streaming Data Pipelines & Real-Time ML Infrastructure: exercises

Advanced Exercise 1: Streaming vs Batch Feature Classification

Context: ShopSense is designing the feature set for the next version of its recommendation engine. The following 8 features are under consideration. Your task is to classify each and design its implementation.

For each feature, provide:

1. Freshness class (historical / batch-fresh / near-real-time / real-time)
2. Justified streaming vs batch decision (with freshness-accuracy reasoning)
3. If streaming: window type and size, watermark strategy, exactly-once vs at-least-once
4. If batch: schedule, storage format, staleness upper bound

Features:

1. user_lifetime_purchase_count: total purchases since account creation
2. product_view_count_last_2hr: number of users who viewed a product in the last 2 hours (used for trending)
3. user_preferred_brands_90d: user's top 5 brands by purchase frequency, last 90 days
4. user_session_view_count: number of product views in the current active session
5. category_conversion_rate_7d: purchase/view ratio per category, last 7 days
6. user_tx_velocity_5min: transaction count in last 5 minutes (fraud feature)
7. product_price_change_indicator: flag set when a product's price dropped > 10% in the last hour
8. user_cart_abandonment_rate_30d: fraction of carts not converted, last 30 days

Deliverable: A feature classification table with implementation justification for each row. For streaming features, include a Flink window configuration snippet.

---

Advanced Exercise 2: Temporal Leakage Diagnosis

Scenario: You are reviewing a colleague's training pipeline for a ShopSense demand forecasting model. The model predicts the probability that a product will be out-of-stock in the next 24 hours. Offline AUC is 0.92, but online precision@K is significantly lower than expected from offline evaluation.

You suspect temporal leakage. Here is the relevant training data assembly code:

def build_training_dataset(events_df: pd.DataFrame, 
                            labels_df: pd.DataFrame) -> pd.DataFrame:
    # Get current feature values for all products in the training set
    product_ids = labels_df["product_id"].unique().tolist()
    
    # Fetch current inventory features
    inventory_features = feature_store.get_online_features(
        features=[
            "inventory_features:current_stock_level",
            "inventory_features:reorder_velocity_7d",
            "inventory_features:supplier_lead_time_days",
            "inventory_features:sales_velocity_24hr",
        ],
        entity_rows=[{"product_id": pid} for pid in product_ids]
    ).to_df()
    
    # Join labels to features
    training_df = labels_df.merge(
        inventory_features, on="product_id", how="left"
    )
    
    # Add event-level features
    event_features = events_df.groupby("product_id").agg({
        "purchase_amount": ["mean", "sum"],
        "view_count": "sum"
    }).reset_index()
    
    return training_df.merge(event_features, on="product_id", how="left")

Your tasks:

1. Identify every source of temporal leakage in this code. Be specific: which line, which feature, why it constitutes leakage.
2. For each identified leakage, estimate the direction and magnitude of the resulting bias (does it inflate or deflate offline AUC?).
3. Write the corrected version of build_training_dataset() using point-in-time correct feature retrieval.
4. Design a temporal leakage CI check that could be added to the training pipeline to automatically detect this class of error.

---

Advanced Exercise 3: Watermark Calibration Analysis

Scenario: ShopSense has deployed a Flink job computing user_session_view_count, the number of product views in the current 30-minute tumbling window. The job uses an out-of-orderness bound of 15 seconds.

During a post-incident review, you discover:

• 8% of mobile app events arrive 20–45 seconds after their event timestamp (mobile batching)
• 2% of events arrive 60–180 seconds late (poor connectivity users on mobile data)
• 0.3% of events arrive > 5 minutes late (background sync from offline mode)
• The recommendation model uses this feature to detect "active session" users (defined as > 3 views in the window)

Your tasks:

1. Estimate the false negative rate for "active session" detection given the current 15-second watermark. Which user cohorts are most affected?
2. Calculate the end-to-end feature latency (event occurrence → feature available in online store) for three watermark settings: 15s, 60s, 3 minutes.
3. Recommend a watermark setting and justify it using the business cost of false negatives vs latency tradeoff.
4. Design a side output handler for late events that arrives > 3 minutes late. What should be done with these events? (Options: drop, recompute window, emit correction to online store, log to monitoring only.)

---

Real-World Implementation 1: Flink Streaming Feature Pipeline

Task: Build a production-ready Flink streaming feature pipeline for ShopSense transaction velocity features.

Scaffold to implement:

# shopsense_streaming_features/transaction_velocity_pipeline.py
from pyflink.datastream import StreamExecutionEnvironment, CheckpointingMode
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common import WatermarkStrategy, Duration, Time
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.datastream.functions import AggregateFunction
import redis
import json


class TransactionEvent:
    """Parsed transaction event with event-time semantics."""
    def __init__(self, user_id: str, transaction_id: str, amount: float,
                 status: str, event_timestamp_ms: int):
        self.user_id = user_id
        self.transaction_id = transaction_id
        self.amount = amount
        self.status = status
        self.event_timestamp_ms = event_timestamp_ms


class VelocityAggregator(AggregateFunction):
    """
    Aggregate: count and sum of completed transactions in window.
    Must handle exactly-once (COUNT is not idempotent).
    """
    def create_accumulator(self):
        # TODO: return initial accumulator state (count=0, amount_sum=0.0)
        pass

    def add(self, value: TransactionEvent, accumulator):
        # TODO: update accumulator with new event (filter status='COMPLETED')
        # and return the updated accumulator (the tests below use the return value)
        pass

    def get_result(self, accumulator) -> dict:
        # TODO: return {"tx_count": int, "tx_amount_sum": float}
        pass

    def merge(self, acc1, acc2):
        # TODO: merge two accumulators (for session windows)
        pass


class DualWriteSink:
    """
    Write velocity features to Redis (online serving) and
    emit records for Delta Lake offline store (via Kafka sink topic).
    """
    def __init__(self, redis_host: str, redis_port: int = 6379):
        self.redis_host = redis_host
        self.redis_port = redis_port
        self._redis_client = None

    def _get_redis(self):
        if self._redis_client is None:
            self._redis_client = redis.Redis(
                host=self.redis_host, port=self.redis_port,
                decode_responses=True
            )
        return self._redis_client

    def write_online(self, user_id: str, window_start_ms: int,
                     tx_count: int, tx_amount_sum: float,
                     ttl_seconds: int = 3600):
        # TODO: write to Redis with key pattern "user:{user_id}:tx_velocity:5min"
        # Store as hash: {"count": tx_count, "amount_sum": tx_amount_sum,
        #                  "window_start": window_start_ms, "updated_at": now}
        pass

    def write_offline_record(self, record: dict) -> dict:
        # TODO: return a record dict for offline Delta Lake logging
        # Must include: user_id, window_start_ms, window_end_ms,
        #               tx_count, tx_amount_sum, computed_at_ms
        pass


def build_velocity_pipeline(
    kafka_bootstrap_servers: str,
    kafka_topic: str,
    redis_host: str,
    window_minutes: int = 5,
    max_out_of_orderness_seconds: int = 30,
    checkpoint_interval_ms: int = 60_000,
) -> StreamExecutionEnvironment:
    """
    Build and return a configured Flink StreamExecutionEnvironment
    for transaction velocity feature computation.
    
    Pipeline: Kafka → parse → filter COMPLETED → watermark → 
              5-min tumbling window → aggregate → DualWriteSink
    """
    env = StreamExecutionEnvironment.get_execution_environment()
    
    # TODO 1: Configure checkpointing (exactly-once, 60s interval,
    #          30s min pause, RocksDB state backend, S3 checkpoint storage)
    
    # TODO 2: Configure Kafka source with:
    #   - EARLIEST offset for new jobs (or COMMITTED for restarts)
    #   - Bounded out-of-orderness watermark strategy using event_timestamp_ms field
    #   - max_out_of_orderness_seconds parameter

    # TODO 3: Build the pipeline:
    #   - Parse JSON → TransactionEvent (handle malformed records → DLQ side output)
    #   - Filter: status == 'COMPLETED' and amount > 0
    #   - Key by user_id
    #   - Apply tumbling event-time window of window_minutes
    #   - Aggregate with VelocityAggregator
    #   - Sink to DualWriteSink (online Redis + offline record emission)

    return env


def run_backfill(
    s3_historical_path: str,
    start_date: str,
    end_date: str,
    redis_host: str,
    write_online: bool = False,  # False for backfill — offline only
) -> None:
    """
    Backfill transaction velocity features from S3 historical event logs.
    Uses bounded source; same aggregation logic as live pipeline.
    """
    # TODO: implement using Flink FileSource (bounded) reading Parquet from S3
    # Key difference from live pipeline:
    #   1. Source is FileSource, not KafkaSource
    #   2. write_online=False: skip Redis write, write offline record only
    #   3. Job terminates when source is exhausted (bounded input)
    pass

Implementation requirements:

1. VelocityAggregator: implement all four methods. add() must be idempotent-safe given exactly-once guarantees.
2. DualWriteSink.write_online(): use Redis HSET with TTL. Include window_start_ms in the stored hash for debugging.
3. build_velocity_pipeline(): configure checkpointing correctly, configure watermark with the event_timestamp_ms field from the JSON payload, handle malformed records via side output to a dead_letter topic.
4. run_backfill(): implement using Flink's bounded FileSource reading the historical Parquet files. Must not write to Redis online store.

Test to pass:

def test_velocity_aggregator_exactly_once():
    agg = VelocityAggregator()
    acc = agg.create_accumulator()
    
    # Process 3 events
    for i in range(3):
        event = TransactionEvent(
            user_id="u1", transaction_id=f"tx{i}",
            amount=10.0 + i, status="COMPLETED",
            event_timestamp_ms=1700000000000 + i * 1000
        )
        acc = agg.add(event, acc)
    
    result = agg.get_result(acc)
    assert result["tx_count"] == 3
    assert abs(result["tx_amount_sum"] - 33.0) < 0.001

def test_aggregator_filters_non_completed():
    agg = VelocityAggregator()
    acc = agg.create_accumulator()
    
    events = [
        TransactionEvent("u1", "tx1", 50.0, "COMPLETED", 1700000000000),
        TransactionEvent("u1", "tx2", 30.0, "PENDING", 1700000001000),
        TransactionEvent("u1", "tx3", 20.0, "FAILED", 1700000002000),
    ]
    for e in events:
        acc = agg.add(e, acc)
    
    result = agg.get_result(acc)
    assert result["tx_count"] == 1
    assert abs(result["tx_amount_sum"] - 50.0) < 0.001

---

Real-World Implementation 2: Point-in-Time Correctness Validator

Task: Build a CI tool that validates point-in-time correctness for any training dataset.

Scaffold:

# shopsense_mlops/pit_correctness_validator.py
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import Optional
from feast import FeatureStore


@dataclass
class SkewReport:
    feature_name: str
    sample_size: int
    pct_rows_with_skew: float        # fraction with > threshold_pct shift
    mean_absolute_shift_pct: float   # average shift across all rows
    max_shift_pct: float             # worst-case row
    has_temporal_leakage: bool       # True if pct_rows_with_skew > leakage_threshold
    direction: str                   # "inflated" | "deflated" | "mixed" | "clean"
    p_value: float                   # KS test p-value


def sample_training_dataset(
    training_df: pd.DataFrame,
    n_samples: int = 1000,
    random_state: int = 42
) -> pd.DataFrame:
    """Sample rows for validation. Preserves temporal distribution."""
    # TODO: stratified sample by date (to avoid over-sampling recent data)
    # Return sample with columns: entity_id, event_timestamp, feature_name, training_value
    pass


def query_historical_features(
    sample_df: pd.DataFrame,
    feature_store: FeatureStore,
    feature_refs: list[str],
) -> pd.DataFrame:
    """
    Query feature store for historical values at each row's event_timestamp.
    Returns sample_df with additional columns: {feature_name}_historical
    """
    # TODO: use feature_store.get_historical_features() with entity_df
    # entity_df must have: entity_id column + event_timestamp column
    # Merge historical values back to sample_df
    pass


def compute_skew_report(
    sample_df: pd.DataFrame,
    feature_name: str,
    training_col: str,
    historical_col: str,
    shift_threshold_pct: float = 0.10,
    leakage_threshold: float = 0.10,
) -> SkewReport:
    """
    Compare training values to point-in-time historical values.
    
    A positive mean shift (training > historical) indicates temporal leakage:
    the model was trained on values that reflect future state.
    """
    # TODO: compute shift per row: |training - historical| / (|historical| + 1e-9)
    # Determine direction: if training_mean > historical_mean → inflated (likely leakage)
    # Run KS test between training values and historical values
    # Return SkewReport dataclass
    pass


def validate_training_dataset(
    training_df: pd.DataFrame,
    feature_store: FeatureStore,
    feature_refs: list[str],
    entity_col: str,
    timestamp_col: str,
    n_samples: int = 1000,
    fail_on_leakage: bool = True,
) -> tuple[bool, list[SkewReport]]:
    """
    Run full point-in-time correctness validation.
    
    Returns (passed, reports). If fail_on_leakage=True and any feature
    has has_temporal_leakage=True, raises ValueError with a summary report.
    
    Usage in training pipeline:
        passed, reports = validate_training_dataset(
            training_df=df,
            feature_store=store,
            feature_refs=["tx_features:tx_count_5min", "tx_features:tx_amount_avg"],
            entity_col="user_id",
            timestamp_col="event_timestamp",
        )
    """
    # TODO: orchestrate sample → query → compute_skew_report for each feature
    # Print a summary table (feature | pct_skewed | mean_shift | leakage?)
    # Raise ValueError if fail_on_leakage=True and any report.has_temporal_leakage=True
    pass


# pytest test
def test_clean_dataset_passes_validation():
    """Training data with correct point-in-time retrieval should pass."""
    # Build a synthetic training dataset where training values == historical values
    # (no leakage). validate_training_dataset should return passed=True.
    pass

def test_leaked_dataset_fails_validation():
    """Training data with future-leaked features should fail validation."""
    # Build synthetic training dataset where training values reflect future state.
    # validate_training_dataset should return passed=False and
    # the leaking feature's SkewReport.has_temporal_leakage should be True.
    pass

Implementation requirements:

1. sample_training_dataset(): stratified by event_timestamp.dt.date; sample proportionally from each date to avoid over-representing recent data.
2. compute_skew_report(): direction detection: if training_mean > historical_mean by > threshold, classify as "inflated" (most likely temporal leakage); if training_mean < historical_mean, "deflated" (possible feature freshness issue at serving time).
3. validate_training_dataset(): print a formatted summary table; the raised ValueError must include the feature name, pct_skewed, and mean_shift_pct to make the error actionable.

---

Production Challenge 1: Streaming Pipeline Cascade Failure

Scenario: ShopSense's Kafka consumer lag monitor fires at 3:47 AM on a Tuesday. The transaction_events topic has 8 million unprocessed messages (normally < 50,000). The Flink job is running but processing at 10% of normal throughput.

Investigation reveals:

• A new product category was added at 3:12 AM, and products were bulk-ingested, 500,000 product_catalog_update events in 8 minutes
• The Flink job performs a stream-stream join: transaction_events JOIN product_catalog_updates ON product_id with a 30-minute event-time window
• The stream-stream join maintains state for all unmatched events within the join window
• The 500,000 catalog updates are all for products from a single new category: category_id = "FLASH_SALE_ELECTRONICS"
• The fraud scoring model is using stale features during this period
• The recommendation model is receiving null for the product_category feature for all transactions in the last 40 minutes

Your tasks:

1. Diagnose the root cause precisely. What is happening in the Flink job state? Why is throughput degraded by 90%?
2. Immediate remediation: what action do you take right now to restore fraud scoring to working state?
3. Architectural fix: how do you redesign the stream-stream join to prevent this from happening again?
4. Monitoring design: what alert should have fired before the 3:47 AM cascade failure? What metric, what threshold, what lead time?
5. Null-feature handling: the recommendation model received null product_category for 40 minutes. What should a well-designed serving system do when a streaming feature is null due to pipeline lag? Design the fallback policy.

---

Production Challenge 2: Training-Serving Skew Post-Deployment Investigation

Scenario: ShopSense deployed a new fraud detection model 6 weeks ago. Offline AUC was 0.91. Online precision is 0.73 (vs 0.79 for the previous model). The model is flagging 40% more transactions as potentially fraudulent, but the false positive rate is higher than expected.

Feature analysis shows:

• tx_velocity_5min (streaming): online serving mean = 1.8, training distribution mean = 3.4
• user_account_age_days (batch): online serving mean = 187, training distribution mean = 187 (✓)
• tx_amount_avg_1hr (streaming): online serving mean = $47.20, training distribution mean = $89.40

Your tasks:

1. Given the feature distribution data, formulate three hypotheses for why tx_velocity_5min shows a 1.6× mean difference between training and serving.
2. Design a diagnostic query to determine which hypothesis is correct. What data do you need, and where would you find it?
3. The tx_amount_avg_1hr shows a 1.9× difference. What is the most likely cause, and how does this differ from the likely cause of the tx_velocity_5min skew?
4. Write a post-incident retrospective section: "What we would do differently in the training pipeline design to catch this before deployment." Include at least 3 concrete process changes.
5. Immediate action: the model is live and producing high false positives. Do you roll back immediately, or keep it running? Justify your decision using a risk framework (false positive cost vs rollback cost vs investigation time).

---

Interview Question 1: Streaming vs Batch: The Architectural Decision

Question (as asked at senior/staff ML engineer interviews):

"We're building a real-time recommendation engine for a flash sale feature that runs for 4 hours, twice per week. During a flash sale, product popularity changes every 5 minutes. Outside flash sales, recommendations are weekly-refresh batch predictions. How would you architect the data pipeline for the trending features? Walk me through your reasoning."

Model answer structure:

Address the key dimensions in order:

1. Freshness requirement analysis. During flash sales, a product's trending score from 5 minutes ago is significantly stale: a product going viral in the last 5 minutes won't surface without near-real-time data. Outside flash sales, weekly-refresh batch is appropriate. The feature has two modes.

2. Architecture recommendation. A hybrid: maintain a streaming pipeline for trending features that runs 24/7 (not just during flash sales), but the batch pipeline writes to a different feature key prefix. The recommendation engine uses the streaming score when freshness is < 10 minutes; falls back to the batch score when streaming is unavailable.

3. Streaming tax justification. A 4-hour event window that is critical to the business justifies streaming infrastructure. The incremental cost of running Flink continuously for this feature set is low relative to the business value. The alternative (spinning up streaming infrastructure only during flash sales) has higher operational risk and cold-start latency.

4. Event-time semantics. Trending score is a COUNT of views. It must be computed over event-time windows. A processing-time window during a spike would produce incorrect counts if Kafka lag increases. Watermark: 30 seconds (product catalog pages are primarily desktop/fast connections).

5. Training data implications. The training pipeline must use point-in-time correct features for flash sale labels, the feature value at the time of the interaction, not the final trending score after the sale ended (which would be the maximum, constituting temporal leakage).

6. Monitoring. Feature freshness SLA: < 30 seconds event-to-serving during flash sales. Alert if freshness degrades > 2 minutes.

---

Interview Question 2: Debugging a Silent Streaming Failure

Question: "Your fraud model's precision has dropped 15% over the last 3 days. The Flink job shows no errors. Kafka consumer lag is normal (< 10,000). Feature freshness monitoring shows < 15 seconds. Walk me through your debugging process."

Model answer structure:

The key insight: all the streaming infrastructure metrics are healthy, which means the failure is likely in feature correctness (wrong values, not missing values) or model distribution shift.

Step 1: Disambiguate pipeline vs model. Pull a sample of recent fraud detections and compare feature values in the model log to what the feature store says those values should be. If they match, the pipeline is correct; the issue is model quality. If they differ, the pipeline is producing wrong values.

Step 2: If pipeline is producing wrong values. Check for recent deployments or schema changes in the last 3 days. A type change in an upstream event (int → float, or a null handling change) can silently alter aggregation values without causing errors. Check the Schema Registry for recent schema version changes.

Step 3: Temporal analysis. When exactly did precision start dropping? If it correlates with a deployment, that's the culprit. If it correlates with a gradual drift, it's more likely distribution shift.

Step 4: Feature distribution comparison. Run detect_feature_skew() comparing the last 3 days of serving feature values to the training distribution. A 1.5× shift in tx_velocity_5min or tx_amount_avg_1hr with no error logs suggests a silent schema change or windowing change.

Step 5: Verify exactly-once. Check checkpoint success metrics. If checkpoints have been failing or taking longer than the checkpoint interval, exactly-once guarantees are void: COUNT aggregations may be double-counting after job restarts.

Output: "My hypothesis is X. Here is the query/check that would confirm or refute it. If confirmed, the fix is Y. If not, my next hypothesis is Z."

---

Interview Question 3: Designing Feature Freshness SLAs

Question: "You're the tech lead for a new streaming ML platform at a mid-size fintech. You need to define feature freshness SLAs for the platform. How do you approach it, and what SLAs would you propose?"

Model answer structure:

Start with business impact, not technology capability. For each feature type, ask: what is the cost of a feature being X minutes stale? For fraud detection velocity features, a 5-minute stale feature can miss 40+ micro-transactions in a velocity attack (direct financial loss). For recommendation trending features, a 30-minute stale feature produces suboptimal recommendations (lower CTR, bounded revenue impact). For user preference features (batch), daily freshness is sufficient.

Differentiate SLA tiers:

• Real-time (fraud, high-stakes): event-to-serving < 10 seconds. Requires streaming compute, Redis online store.
• Near-real-time (recommendations, session): event-to-serving < 60 seconds. Streaming compute, Redis or Cassandra.
• Batch-fresh (user profiles, long-term preferences): updated within 2 hours of batch job completion. Nightly batch, offline feature store.
• Historical (demographics, account metadata): weekly refresh acceptable. Pure batch.

Translate SLAs to architecture requirements:

• < 10 seconds requires Kafka → Flink with short checkpoint intervals (30s), Redis with < 5ms get latency, direct serving path
• < 60 seconds can tolerate 60s checkpoint intervals, allows for Cassandra (higher latency but cheaper at scale)

• 5 minutes can be handled with micro-batch (Spark Structured Streaming with 5-minute trigger)

Monitoring: For each SLA tier, define a P95 freshness metric and a breach SLA threshold. Alert at 2× SLA for investigation, page at 5× SLA for immediate response.

The trap to avoid: Don't define SLAs in terms of what the infrastructure can do today. Define them in terms of what the business needs, then design the infrastructure to meet them. Teams that do it backwards end up with SLAs that are technically achievable but commercially meaningless.