Why sampling fails on skewed telemetry

Netflix's member-sessionizer consolidates tens of billions of hourly client events — taps, playbacks, engagement signals — into per-profile viewing sessions. When the pipeline moved from a legacy logging architecture to a modern platform, the new job produced executor OOMs, heavy shuffle spill, and thousands of stage retries.

None of it reproduced on sampled data. The cause was skew: an idle tab left streaming overnight emits tens of thousands of telemetry events, while a typical session emits only a handful. A 1% sample misses those long-tail sessions entirely, so every test had to run against a full hour of unsampled production data — hours per attempt, three or four attempts per day.

One branch, one experiment

The unlock was a feature of Dataflow, Netflix's command-line tool for building, testing and deploying pipelines: branch-driven development. Workflows run on Maestro, Netflix's orchestrator, and already incorporate a sanitized branch name via the ${dataflow.sanitizedbranch} placeholder. Pushing a branch such as pgn/cap-experiment tags the deployed workflow, its tables and its run namespace with that branch name, so runs never collide with teammates or with each other.

With isolation in place, the model became: one branch is one independent experiment. Scheduler parameters — cap size, shuffle partitions, heavy-device threshold, feature flags — were parameterized per branch, each variant writing to its own output table. Five variants launched simultaneously against the same time window, so total wall-clock time collapsed from the sum of all runs to the duration of the slowest one. Table management cost nothing: Dataflow provisions and updates tables from declared schemas, so schema changes between experiments required no CREATE TABLE or ALTER TABLE.

Cluster contention was the one limit. These jobs share resources with the rest of the company, so at peak only two concurrent runs were sometimes possible instead of five — still a large gain over serial debugging.

The principle: don't optimize a single multi-hour run, build a loop where many full-scale runs execute in parallel without interference.

Agents for operations, tmux for persistence

Tracking instance IDs, configs, memory spills and stage failures across five variants gets messy quickly. An AI agent handled that operational grunt work: variants were described in plain English, and the agent updated scheduler configs, deployed to the branch, triggered jobs and polled metrics, then compared spill and failure counts in a summary table — answering questions like "which run is closest to OOM-free at peak load?" directly. Its value was operational memory, not Spark expertise.

The workflow was eventually packaged as a reusable agent skill at .agents/skills/sandbox-deploy-and-run/, invoked with /sandbox-deploy-and-run <module> <scheduler-file> [PARAM=value]. It covers preflight, deploy, background polling, auto-diagnostics and re-entrant resume <instance_id>, reducing a multi-hour experiment to a single command.

Everything ran on Workbench, Netflix's cloud dev environment, with the laptop as a thin terminal. Without session persistence an SSH drop or a sleeping laptop would kill a multi-hour Spark job, so each experiment lived in a named tmux session on Workbench and could be reattached hours later. The current default setup is a local Ghostty terminal, a dedicated git worktree per branch, and persistent tmux sessions on Workbench — the laptop is disposable. Batches launched at the gate could be checked over inflight Wi-Fi, fitting 4–5 full experiment cycles into a single 24-hour flight.

Row width, not data volume

The root cause traced back to one operator: collect_list, which buffers each full event bag in memory. Long-session OOMs were driven by individual bag size rather than total dataset volume — the legacy pipeline handled identical row counts without crashing, which showed that each event in the new bag had grown considerably heavier.

Two factors inflated event weight:

  • Field types and Option boxing: the session ID moved from an 8-byte Long to a UUID string. Worse, wrapping fields in Scala Option forced Spark to deserialize typed Datasets into JVM objects (Some wrappers), raising object allocation and GC pressure across tens of thousands of events per session.
  • Bloated string attributes: low-value fields such as referrer URLs and test allocations added several hundred bytes per event. Multiplied across collect_list, that wasted terabytes per hour even though downstream tasks rarely read it.

Two fixes followed: removing Option wrappers from non-nullable fields dropped peak executor memory from 112% to 58% of the slot with zero disk spill, and slimming the event bag by dropping unused per-event attributes and hoisting session-invariant metadata (app version, country) to the session header.

Two surprises in join planning

  • Narrower rows caused faster OOMs. Slimming the table pushed its size below Spark's auto-broadcast threshold, so the planner switched from shuffle joins to broadcast joins, which pin build-side tables directly into executor memory. The slimmer rows increased total memory pressure.
  • Broadcast hash joins drained memory silently. Adaptive Query Execution rewrote nine sort-merge joins into broadcast hash joins at runtime. Each stayed under the 400 MB broadcast limit in isolation, but AQE pinned ~3.1 GB of cumulative data simultaneously in a single executor, pushing containers over the limit.

The fix was surgical rather than global: lowering only AQE's runtime threshold, spark.sql.adaptive.autoBroadcastJoinThreshold, to 10 MB while leaving the compile-time threshold alone. That freed the ~3.1 GB without affecting initial query planning. Broadcast joins accelerate queries, but pinned memory is not free — watch accumulated broadcasts in the execution plan, not just dataset sizes.

Discipline enforced by a log

With dozens of parallel tests, the real risk is losing track of what was tried and which JAR and config produced a result. The agent was scaffolded to append a Markdown table row after every run, with columns: Run # | Jar ver | Change vs. Previous Run | Config | Key Metrics (Spill, Failed Tasks, Peak Memory, Duration). Key findings got a short note on what the run proved or disproved.

The log gave the investigation an auditable single source of truth, but its second effect mattered more: requiring a "Change vs. Previous Run" entry forced one variable at a time. Bundling changes made root causes unisolatable and the test had to be rerun — a mistake the format exposes immediately. A secondary document, events-join-spark-tuning.md, distilled the findings into a symptom-to-knob lookup so teammates could skip the chronological history.

The fix that only worked off-peak

To cap session size, row_number().over(Window...) was added and the first N events filtered. It passed off-peak tests and OOM'd at peak. The problem was structural: Spark's Window operator buffers an entire session in memory before evaluating row numbers, so the cap trimmed output downstream without limiting memory during buffering, and one skewed session overwhelmed the executor. Truncation was moved ahead of the Window stage to cap events before buffering. Fixes have to be tested against worst-case skew; off-peak runs hide structural bottlenecks that peak scale triggers immediately.

What shipped

  • Unboxed never-null fields from Option to plain primitives. Peak executor memory fell from about 112% of the slot to about 58%, with zero spill.
  • Slimmed the event bag. Per-event fields nobody used were dropped, and session-invariant fields (versions, country, connection type) were hoisted to session level, stored once per session rather than per event.
  • Capped session size before the Window stage. Oversized sessions are dropped ahead of buffering, and the bag is truncated to the cap rather than whole sessions being dropped.
  • Lowered AQE's runtime broadcast threshold to 10 MB via spark.sql.adaptive.autoBroadcastJoinThreshold, leaving the compile-time threshold untouched.

Each change was isolated to a single experiment before it landed.

The ceiling that remains

These fixes let the pipeline run cleanly, but they did not address the underlying scaling limit: aggregating a full session's event bag into one row has a hard ceiling. As the upstream telemetry source adds attributes, every event in the bag gets heavier, and the memory pressure returns without a single line of code changing on our side. Slimming and unboxing bought headroom; they did not change the data shape. Tuning buys time, not infinite scale — if the data model permits unbounded row growth, the architecture eventually has to change.

Practices worth keeping

  • When a bug only appears at scale, stop sampling. Run many full-scale runs in parallel.
  • If the scheduler can use the branch name, one branch becomes one isolated experiment and parallelism becomes easy.
  • Use an AI agent for operations, not just code: deploy, poll, pull metrics, compare variants — that is where the time goes.
  • Package a repeated loop as a skill, turning the deploy-run-watch-diagnose cycle into one command.
  • Use tmux for long runs; the run must outlive a closed laptop.
  • Keep one experiment log where every entry states the single change from the previous run. The format enforces changing one variable at a time.
  • For memory problems, inspect per-row width before total data volume — bigger field types, Option boxing and fat unused strings all add up inside a collect_list bag.
  • Tuning buys headroom, not a new ceiling. Unbounded data shapes eventually require design changes.