Working with Large-Scale Datasets Questions

Requirements & scope: monitor per-feature and joint drift between training distribution and live inputs for tabular features, detect meaningful shifts while avoiding alert fatigue, provide human-first alerts and optional automated remediation for high-confidence failures.

Metrics

• Univariate: Population Stability Index (PSI) per numeric/categorical feature (binned), Kolmogorov–Smirnov (KS) for continuous, Chi-squared for categorical. Use Jensen–Shannon (symmetric, finite) or KL divergence where distributions smooth.
• Multivariate: Covariate shift via Maximum Mean Discrepancy (MMD) or classifier two-sample test (train a logistic regressor to distinguish train vs prod; AUC > 0.6 indicates drift).
• Complement with simple stats: mean, std, skew, missing rate, unique counts, cardinality.

Sampling cadence & windows

• Real-time low-latency: stream summary stats every 5–15 minutes for critical systems.
• Batch aggregation: daily and weekly windows; maintain rolling window (e.g., last 7/30 days) vs baseline (training or recent stable period).
• Minimum sample: require Nmin per feature (e.g., 500–2000 rows) before computing statistical metrics to avoid noise.

Thresholds & alerting behavior

• Per-feature thresholds (example starting points):
  • PSI: <0.1 OK, 0.1–0.25 warning, >0.25 critical.
  • KS p-value <0.01 for significant difference.
  • Jensen–Shannon divergence: tuned per dataset; initial heuristic 0.1–0.2.
  • Classifier AUC: <0.55 marginal, >0.65 strong drift.
• Alerting policy:
  • Warning: single metric crosses warning threshold — create ticket, notify on-call ML engineer.
  • Critical: persistent breach across >k windows (e.g., 3 consecutive daily windows) or multiple correlated features drift — page ops and stakeholders.
  • Correlation with model performance: if drift coincides with drop in key business metrics (e.g., +5% error) escalate immediately.

Actions: alert-first then escalate

• Alert-only by default: human reviews, run diagnostics (feature importance change, partial dependence).
• Automated mitigation chain (opt-in, gated):
  1. Canary traffic reduction (if model serving supports): route traffic to fallback model.
  2. Auto-retrain trigger: only if drift validated by classifier test + sustained for threshold window and model performance degraded; retrain using recent labeled data with human-in-loop approval.
  3. Rollback: automatic rollback only for severe, high-confidence production-impacting drift (e.g., multiple metrics critical + performance drop), with automatic quarantine and manual post-mortem required.

Tuning thresholds to avoid false positives

• Baseline calibration: compute metric distributions on historical stable periods and set thresholds at, e.g., 95th percentiles rather than fixed heuristics.
• Use holdout validation: simulate drift scenarios and run backtests to measure precision/recall of alerts; choose thresholds maximizing F1 for actual harmful drift.
• Require multiple signals: only escalate when at least 2 orthogonal metrics (e.g., PSI + classifier AUC) or persistence across windows.
• Adaptive thresholds: increase sensitivity during known seasonality adjustments; employ seasonally-aware baselines.
• Rate-limit alerts and use alert deduplication and grouping by root-cause feature clusters.

Diagnostics & observability

• Bubble up context: sample distributions, delta histograms, top-affected features, model SHAP/feature importance, recent data pipeline changes.
• Logging: store all drift scores for trend analysis and threshold re-tuning.

This design emphasizes conservative automation, multi-signal confirmation, and iterative threshold tuning against historical behavior to minimize false positives while ensuring timely detection.

Approach (high level):

• 1st pass: compute cheap, fast fingerprints without loading whole blob — use file size + sample-based fast hash (xxhash64) over a few fixed positions (start, middle, end) to quickly group likely duplicates.
• 2nd pass: for groups with matching cheap fingerprints, compute a full cryptographic hash (SHA-256) by streaming the file in chunks (e.g., 4MB) to avoid loading >1MB fully.
• 3rd (only if needed): if SHA-256 also matches (extremely rare collision), do a final streaming byte-by-byte comparison between two files to guarantee equality.
  This minimizes I/O and memory and avoids false positives.

Code sketch (python):

import os, hashlib
import xxhash

CHUNK = 4 * 1024 * 1024  # 4MB
SAMPLES = [0, 65536, -65536]  # offsets: start, 64KB, 64KB from end

def sample_hash(path):
    size = os.path.getsize(path)
    h = xxhash.xxh64()
    h.update(str(size).encode())
    with open(path, 'rb') as f:
        for off in SAMPLES:
            if off >= 0:
                f.seek(min(off, max(size-1,0)))
            else:
                f.seek(max(size + off, 0))
            chunk = f.read(8192)  # small read
            h.update(chunk)
    return h.intdigest()

def full_sha256(path):
    sha = hashlib.sha256()
    with open(path, 'rb') as f:
        while True:
            b = f.read(CHUNK)
            if not b: break
            sha.update(b)
    return sha.hexdigest()

def bytes_equal(a, b):
    with open(a,'rb') as fa, open(b,'rb') as fb:
        while True:
            ba = fa.read(CHUNK)
            bb = fb.read(CHUNK)
            if not ba and not bb: return True
            if ba != bb: return False

def deduplicate(paths):
    # 1. cheap fingerprinting
    cheap = {}
    for p in paths:
        k = sample_hash(p)
        cheap.setdefault((os.path.getsize(p), k), []).append(p)

    uniques = []
    for group in cheap.values():
        if len(group) == 1:
            uniques.append(group[0]); continue
        # 2. full hash for group members
        by_full = {}
        for p in group:
            h = full_sha256(p)
            by_full.setdefault(h, []).append(p)
        # 3. final verification if multiple same sha
        for h, members in by_full.items():
            rep = members[0]
            uniques.append(rep)
            for other in members[1:]:
                if not bytes_equal(rep, other):
                    uniques.append(other)  # different despite sha (extremely rare)
    return uniques

Key design choices & justification:

• Sample-based xxhash64 + size: extremely fast, small reads, groups likely duplicates while keeping FP low.
• SHA-256 streaming: cryptographically strong, computed incrementally to avoid memory pressure.
• Final streaming byte-compare: absolute equality guarantee if hashes collide.
• Chunked I/O (4MB): balances syscalls and memory; adjust to environment.
  Complexity:
• I/O: in best case (few collisions), reads only samples + full reads for actual duplicates; worst-case you stream every file once O(total_bytes).
• Memory: O(1) per file (streaming buffers); metadata maps sized by number of files.
  Edge cases:
• Very small files (handles via min offsets), permission errors (catch exceptions in production), concurrent file changes (use file locks or snapshot).

Stratified sampling splits data so each subset (train/validation) preserves the original distribution of one or more categorical variables (strata). It’s important when labels are imbalanced or rare — otherwise the validation set may miss rare classes and give misleading metrics.

Concrete example:

• Binary churn label where positive churn = 0.1% of 50M users. A random 1% validation sample might contain few or zero churn examples, so you must stratify by churn to ensure the validation set has representative positives for reliable evaluation.

scikit-learn (if dataset fits memory):

from sklearn.model_selection import train_test_split

X_train, X_val, y_train, y_val = train_test_split(
    X, y, test_size=0.01, random_state=42, stratify=y
)

For very large data (50M rows) that cannot be loaded into memory, do stratified sampling in the data store (SQL) and materialize sampled subsets. Example SQL (Postgres / BigQuery style) that selects proportional samples per label:

WITH counts AS (
  SELECT label, COUNT(*) AS cnt FROM table GROUP BY label
),
targets AS (
  SELECT label, CAST(ROUND(cnt * 0.01) AS INT) AS take
  FROM counts
)
SELECT t.*
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY label ORDER BY RANDOM()) AS rn
  FROM table
) t
JOIN targets trg USING (label)
WHERE t.rn <= trg.take;

Notes and best practices:

• Choose test/val fractions so rare classes have enough examples (e.g., ensure at least N positives in validation).
• For multi-label or hierarchical strata, stratify on a composite key (or use more advanced sampling like stratified on the most important label).
• If using distributed engines (Spark), use sampleBy or groupby with reservoir sampling to scale.
• Always set random seed for reproducibility.

Feature scaling matters because many ML algorithms assume features are on comparable scales; unscaled features can bias distance metrics, gradient updates, regularization, and convergence speed.

Definitions & differences:

• Standardization (z-score): x' = (x - mean) / std. Centers to zero mean and unit variance. Good when data is approximately Gaussian and you care about variance-based weighting.
• Min–max normalization: x' = (x - min) / (max - min). Maps features to [0,1] (or any range). Preserves relative ordering and bounds; sensitive to outliers.
• Robust scaling (median & IQR): x' = (x - median) / IQR. Centers using median and scales by interquartile range; resilient to outliers and skewed distributions.

When scaling is critical:

• Distance-based algorithms: k-NN, K-means, DBSCAN — distances directly affected.
• Gradient-based models: logistic regression, neural networks, SVM (with RBF), and models with L1/L2 regularization — scaling stabilizes optimization and ensures regularization applies uniformly.
• Tree-based models (random forests, decision trees, gradient-boosted trees) are largely invariant to monotonic scaling and usually don’t require scaling.

Integrating into a large-scale pipeline:

• Fit scaler on training data (or training shard) only; persist parameters (mean/std/min/max/median/IQR) with model artifact.
• Implement scaling as a reusable, versioned preprocessing component (e.g., TF Transform, scikit-learn Pipeline, Spark ML Transformer) so same logic runs in training and serving.
• For distributed training, compute global statistics via aggregation (map-reduce: compute counts, sums, sums of squares or approximate quantiles) to avoid bias from shards.
• Validate: include unit tests for transforms, monitor feature distributions in production, and handle novel/unseen values (clipping, fallback).
  Choice guideline: use standardization for most continuous features, min–max when bounded ranges are desired (e.g., image pixels), and robust scaling when outliers/skew are present.

Approach: I'll implement a Count-Min Sketch (CMS) with an auxiliary min-heap to track candidate top-k heavy hitters. CMS gives sublinear memory, O(1) update/query, with one-sided overestimation error; the heap maintains the k items with largest estimated counts. I'll include merge (additive) across workers and state error guarantees.

import mmh3
import heapq
from math import ceil, log

class CountMinTopK:
    def __init__(self, eps=0.001, delta=1e-3, k=100):
        # width w = e/eps, depth d = ln(1/delta)
        self.w = ceil(2.7182818 / eps)
        self.d = ceil(log(1/delta))
        self.k = k
        self.tables = [[0]*self.w for _ in range(self.d)]
        self.seeds = [i*0x9e3779b1 for i in range(self.d)]
        self.heap = []               # min-heap of (est_count, item)
        self.pos = {}                # item -> (est_count)
    
    def _index(self, item, i):
        return mmh3.hash(str(item), self.seeds[i]) % self.w

    def update(self, item, count=1):
        # update CMS
        for i in range(self.d):
            self.tables[i][self._index(item, i)] += count
        est = self.query(item)
        # maintain top-k heap
        if item in self.pos:
            self.pos[item] = est
            # rebuild heap lazily when needed
        else:
            if len(self.heap) < self.k:
                heapq.heappush(self.heap, (est, item))
                self.pos[item] = est
            else:
                if est > self.heap[0][0]:
                    # pop smallest
                    small_est, small_item = heapq.heapreplace(self.heap, (est, item))
                    del self.pos[small_item]
                    self.pos[item] = est

    def query(self, item):
        return min(self.tables[i][self._index(item, i)] for i in range(self.d))

    def topk(self):
        # return sorted top-k by estimated count (descending)
        return sorted(self.heap, key=lambda x: -x[0])

    def merge(self, other):
        assert self.w == other.w and self.d == other.d and self.seeds == other.seeds
        for i in range(self.d):
            for j in range(self.w):
                self.tables[i][j] += other.tables[i][j]
        # rebuild heap from union of pos keys or optionally from candidates
        candidates = set(self.pos.keys()) | set(other.pos.keys())
        self.heap = []
        self.pos = {}
        for item in candidates:
            est = self.query(item)
            if len(self.heap) < self.k:
                heapq.heappush(self.heap, (est, item)); self.pos[item]=est
            else:
                if est > self.heap[0][0]:
                    _, small_item = heapq.heapreplace(self.heap, (est, item))
                    del self.pos[small_item]
                    self.pos[item]=est

Key concepts & guarantees:

• Error bound: for parameters w,d, any query overestimates true count f by at most eps * total_count with probability ≥ 1 - delta (one-sided).
• Space: O(w*d) counters. Time: O(d) per update/query.
• Heap introduces false negatives possibility: CMS can overestimate but never underestimate; using heap on estimated counts may miss items whose estimated counts are low due to hash collisions. To reduce misses, increase w (reduce eps) or maintain a small Space-Saving sketch per worker and merge.
• Merge: CMS is linear — element-wise addition of tables; heavy-hitter candidates can be merged by recomputing estimates for union of candidates. This is associative and commutative, suitable for distributed workers.
  Edge cases:
• Very skewed streams → better with Space-Saving (exact top-k in limited memory).
• High cardinality with small k → increase d,w or use hybrid: CMS for counts + Space-Saving for exact top candidates.