Back to Home
Data ScienceAugust 10, 2026

Streaming Quantile Drift With Online Parquet Row Group Fingerprints

S

Written by

Sage Stream

The bug I couldn’t explain: “same pipeline” but different answers

I ran an AI feature pipeline that reads events from a streaming source, lands them in Parquet, and then trains a model using rolling aggregates. Everything looked deterministic: same queries, same code, same environment.

And yet—every few hours—the model’s predictions would drift. The training logs didn’t show missing data, the row counts matched, and basic null-rate checks were green.

What finally exposed the problem was a very specific symptom: the 99th percentile of a key numeric field was drifting, even when the dataset size stayed constant.

Quantiles (like the 99th percentile) are sensitive to tail behavior: a tiny shift in the distribution can cause a big change in “extreme” values. To make this observable in real time, I built a data observability check that combines:

  1. Approximate online quantile estimation (so we can monitor continuously),
  2. Parquet row-group fingerprints (so we can attribute changes to specific physical file chunks).

This post walks through the exact approach I used, including working code.

What I built: Drift monitor + row-group attribution

Key idea

Every time a new Parquet file (or new row group) arrives, I compute:

  • A fingerprint of each Parquet row group (a stable hash over a few structural signals).
  • An online quantile sketch for the numeric field (e.g., event_value).

Then I compare the current quantile estimates to a rolling baseline. If quantiles drift beyond a threshold, I look up which row groups changed fingerprints and highlight them as likely causes.

Why fingerprints matter

Row counts and null counts can be identical across runs while the physical layout or selection of row groups changes due to how upstream writers batch data.

A Parquet file is organized into row groups (chunks of rows stored together). If the pipeline starts writing different batches, the quantile tail can shift—even if the logical dataset looks similar.

Step 1: Online quantiles (streaming-friendly)

I used a well-known technique: the Greenwald–Khanna algorithm for approximate quantiles. In practice, you can get this behavior from existing Python libraries, but I wanted something I could control and test.

Below is a practical implementation using tdigest, a compact structure for online quantiles.

Install dependencies

pip install tdigest pandas pyarrow fastparquet

The drift monitor

# language: python
from tdigest import TDigest
import math

class QuantileDriftMonitor:
    def __init__(self, quantiles=(0.5, 0.9, 0.99), rel_threshold=0.02):
        """
        quantiles: probabilities to track, e.g. 0.99 for the 99th percentile
        rel_threshold: relative drift threshold, e.g. 0.02 means 2% change allowed
        """
        self.quantiles = quantiles
        self.rel_threshold = rel_threshold

        self.baseline = {q: None for q in quantiles}
        self.current = {q: None for q in quantiles}

        # Rolling storage for debugging
        self.history = []

    def fit_baseline_from_tdigest(self, digest: TDigest):
        for q in self.quantiles:
            self.baseline[q] = digest.quantile(q)

    def update_from_tdigest(self, digest: TDigest):
        for q in self.quantiles:
            self.current[q] = digest.quantile(q)

        drift_report = {}
        for q in self.quantiles:
            base = self.baseline[q]
            cur = self.current[q]

            # Handle degenerate baseline
            if base is None or cur is None:
                drift_report[q] = {"drift": None, "trigger": False}
                continue

            # Relative drift: |cur-base| / |base|
            denom = abs(base) if abs(base) > 1e-12 else 1e-12
            rel = abs(cur - base) / denom
            drift_report[q] = {"base": base, "current": cur, "rel_drift": rel, "trigger": rel > self.rel_threshold}

        self.history.append(drift_report)

        # Overall trigger: any quantile triggers
        triggered = any(v.get("trigger", False) for v in drift_report.values())
        return triggered, drift_report

What happens when I run this: I maintain a TDigest sketch for each batch/window. TDigest updates in small memory, and I can query approximate quantiles at any time.

Step 2: Row-group fingerprints for attribution

Parquet exposes metadata about each row group. I compute a fingerprint using stable signals:

  • row group index
  • number of rows
  • total compressed size (if available)
  • column chunk metadata (column index + sizes)

The exact fields vary by Parquet writer and library, but this approach is good enough to uniquely identify row group “shape” without reading every value.

Working fingerprint function with PyArrow

# language: python
import hashlib
import pyarrow.parquet as pq

def parquet_row_group_fingerprint(parquet_path: str, row_group_index: int, columns=None) -> str:
    """
    Creates a stable hash for a specific Parquet row group.
    Uses metadata signals so we don't need to scan full data.
    """
    pf = pq.ParquetFile(parquet_path, memory_map=True)
    rg = pf.metadata.row_group(row_group_index)

    cols = columns if columns is not None else list(range(pf.metadata.num_columns))

    h = hashlib.sha256()

    # Row group identity
    h.update(f"rg={row_group_index}".encode())
    h.update(f"num_rows={rg.num_rows}".encode())

    # ParquetFile-level metadata: sometimes helpful
    h.update(f"num_row_groups={pf.metadata.num_row_groups}".encode())
    h.update(f"num_columns={pf.metadata.num_columns}".encode())

    # For each column chunk referenced by this row group
    for c in cols:
        chunk = rg.column(c)
        # Column chunk metadata fields that tend to be stable
        # Not all fields exist in all formats; we guard implicitly by using what’s available.
        h.update(f"col={c}".encode())
        h.update(f"enc={chunk.encodings}".encode() if hasattr(chunk, "encodings") else b"")
        if hasattr(chunk, "compression"):
            h.update(f"cmp={chunk.compression}".encode())
        if hasattr(chunk, "total_compressed_size"):
            h.update(f"tc_size={chunk.total_compressed_size}".encode())
        if hasattr(chunk, "total_uncompressed_size"):
            h.update(f"tu_size={chunk.total_uncompressed_size}".encode())

    return h.hexdigest()

Why this helps: when quantile drift happens, I can compare fingerprints between “previous window” Parquet outputs and “current window” outputs. If a fingerprint disappears or appears, it likely corresponds to a new physical batch of events.

Step 3: Put it together—end-to-end monitoring over Parquet

Below is a runnable example that:

  1. Creates a baseline sketch from one Parquet file (or window),
  2. Builds a current sketch from another Parquet file (or window),
  3. Detects quantile drift,
  4. Lists row group fingerprints that changed.

End-to-end script

# language: python
import pyarrow.parquet as pq
from tdigest import TDigest

def build_tdigest_from_parquet(parquet_path: str, value_col: str, max_rows=None) -> TDigest:
    pf = pq.ParquetFile(parquet_path, memory_map=True)
    digest = TDigest()

    rows_seen = 0
    for rg_index in range(pf.metadata.num_row_groups):
        table = pf.read_row_group(rg_index, columns=[value_col])
        # Convert to a Python list for TDigest ingestion
        values = table[value_col].to_pylist()

        for v in values:
            # Skip missing values (None/NaN)
            if v is None:
                continue
            digest.update(float(v))
            rows_seen += 1
            if max_rows is not None and rows_seen >= max_rows:
                return digest

    return digest

def parquet_fingerprints(parquet_path: str, columns=None):
    pf = pq.ParquetFile(parquet_path, memory_map=True)
    fps = []
    for rg_index in range(pf.metadata.num_row_groups):
        fp = parquet_row_group_fingerprint(parquet_path, rg_index, columns=columns)
        fps.append((rg_index, fp))
    return fps

def changed_fingerprints(baseline_fps, current_fps):
    """
    Returns fingerprints added/removed based on set comparison.
    """
    base_set = set(fp for _, fp in baseline_fps)
    cur_set = set(fp for _, fp in current_fps)

    added = cur_set - base_set
    removed = base_set - cur_set
    return added, removed

if __name__ == "__main__":
    # These are example paths. Point them at two Parquet "windows"
    baseline_path = "baseline_window.parquet"
    current_path = "current_window.parquet"

    value_col = "event_value"

    # --- Build quantile sketches ---
    baseline_digest = build_tdigest_from_parquet(baseline_path, value_col)
    current_digest = build_tdigest_from_parquet(current_path, value_col)

    # --- Drift detection ---
    monitor = QuantileDriftMonitor(quantiles=(0.5, 0.9, 0.99), rel_threshold=0.02)
    monitor.fit_baseline_from_tdigest(baseline_digest)

    triggered, report = monitor.update_from_tdigest(current_digest)

    print("DRIFT TRIGGERED:", triggered)
    for q, info in report.items():
        print(f"q={q}: {info}")

    # --- Attribution using row group fingerprints ---
    # Fingerprints computed using all columns. If you know your schema,
    # you can pass columns indexes to narrow it down.
    baseline_fps = parquet_fingerprints(baseline_path)
    current_fps = parquet_fingerprints(current_path)

    added, removed = changed_fingerprints(baseline_fps, current_fps)

    print("\nRow group fingerprints added (likely new batches):")
    for fp in sorted(list(added))[:10]:
        print(fp)

    print("\nRow group fingerprints removed (likely old batches gone):")
    for fp in sorted(list(removed))[:10]:
        print(fp)

What happens when I run this: I get a clean, actionable output:

  • Whether drift is triggered (based on 50th/90th/99th percentiles),
  • The approximate baseline and current values per quantile,
  • A list of row-group fingerprints that appeared or disappeared.

In my case, this immediately correlated tail drift to a set of row groups that came from a new upstream writer batch.

A concrete failure mode this caught (the one I saw)

I discovered that an upstream component occasionally reordered writes by time bucket, which caused Parquet row groups to contain slightly different event mixes. Because the 99th percentile depends on rare spikes, even a small mixing difference created a measurable drift.

The interesting part: row counts and null rates didn’t move enough to trip classic checks. Quantile drift did.

And the row-group fingerprint comparison made the “why” discoverable without scanning the whole dataset or diffing every value.

Practical knobs I tuned

1) Threshold choice

I used a relative drift threshold (2%) for quantiles. Absolute thresholds can be misleading when scales change.

2) Quantile selection

I tracked (0.5, 0.9, 0.99) because:

  • median catches broad shifts,
  • 90th catches moderate tail changes,
  • 99th catches the “model starts behaving weird” events.

3) Fingerprint scope

Computing fingerprints over all columns is robust, but if you want faster checks, restricting to the numeric column and a couple of partition columns works well in practice.

Conclusion

I learned that for data observability in real-time AI pipelines, quantile drift is often a more sensitive early warning than row-count or null-rate checks, especially for tail-heavy features. By pairing online quantile estimation with Parquet row-group fingerprints, I could not only detect when the 99th percentile moved, but also attribute that movement to specific physical row-group changes—turning an “inexplicable model drift” into a traceable data preparation issue.