Senior Data Engineer at Netflix - Comprehensive Interview Preparation Guide

Direct answer

The three main ways to split a training job are data parallelism (each worker holds a full copy of the model and a different slice of the data, and workers exchange gradients to stay in sync), model or tensor parallelism (the model itself, or a single large layer, is split across workers, which exchange intermediate activations mid-computation), and pipeline parallelism (the model's layers are grouped into sequential stages across workers, which exchange only the activations at each stage boundary). Which one you reach for depends on whether the bottleneck is dataset size, model size, or both.

Structured elaboration

In practice, training very large models typically combines all three, sometimes called 3D parallelism: tensor parallelism within a machine over a fast local interconnect, pipeline parallelism across machines within a group, and data parallelism across groups, because each strategy's communication pattern fits a different tier of network speed: the fastest links carry the most frequent, tensor-parallel traffic, and the slowest cross-group links carry the least frequent, data-parallel gradient synchronization.

Worked example

Consider an 8-billion-parameter model stored at 16-bit precision (2 bytes per parameter): weights alone need roughly $8\times10^9 \times 2 = 16 \text{ GB}$. A common optimizer keeps additional per-parameter state that can add another 2 to 3x on top of the weights; at 3x, total memory needed is $16 \times 3 = 48$ GB. If each device has 24 GB of memory, this does not fit on one device at all, and no amount of extra data-parallel workers changes that, since every data-parallel worker would still need to hold the full 48 GB. Splitting the model across 2 devices (model or tensor parallelism) brings the per-device share to $48/2 = 24$ GB, right at the edge of fitting; that arithmetic, not preference, is what forces model parallelism here.

Separately, consider the communication cost that makes pure data parallelism expensive at this scale: an all-reduce of an 8-billion-parameter gradient tensor at 2 bytes each moves 16 GB of gradient data across workers every step. At a hypothetical 100 gigabit-per-second (12.5 GB/s) interconnect, transferring that volume alone takes on the order of:

$16 / 12.5 \approx 1.28 \text{ seconds of network time per step, in the naive case}$

Efficient ring all-reduce implementations move closer to $2\times(N-1)/N$ of that volume per node rather than the full amount, but the underlying point holds: gradient-synchronization cost scales with model size, not batch size, which is why pure data parallelism becomes network-bound long before it becomes compute-bound on very large models, and why the other two strategies exist at all.

Trade-offs & pitfalls

• Reaching for model parallelism when the real bottleneck is dataset throughput and the model fits comfortably in memory adds mid-layer communication overhead for no benefit, when data parallelism's simpler gradient-only synchronization would have been enough.
• Pipeline parallelism's bubble problem, workers idling while waiting on earlier or later stages, shrinks but never disappears with more micro-batches; more micro-batches reduce the idle fraction but increase per-batch overhead, so the choice is a tuning trade-off, not a solved problem.
• Data parallelism's communication cost scales with model size, so simply adding more data-parallel workers to speed up training on a very large model eventually stops helping once the job becomes network-bound.
• Combining all three strategies is powerful but adds real engineering and debugging complexity around exactly which workers communicate with which, in what order, at what precision; reaching for that combination before confirming plain data parallelism is actually insufficient is a common overengineering trap.

Pros of surrogate integer keys: smaller storage, faster joins (integer comparisons), stable joins despite source key churn, and compact clustered indexes—important at hundreds of millions of rows. They decouple ETL changes and support SCDs. Cons: extra mapping layer (natural -> surrogate) and need for lookups when reconciling. Natural keys pros: human-readable, no lookup table needed. Cons: larger size, volatility, potential duplicates, schema coupling, and slower joins. Recommendation: use surrogate integer keys as primary keys for dimension tables at scale, keep natural_key as a unique column with an index for traceability, and maintain mapping in ETL. This balances performance, stability, and auditability.

Direct answer

I noticed a recurring, tedious task nobody had gotten around to fixing, built the smallest tool that solved it rather than the most impressive one I could imagine, and ran it alongside the manual process for a while before trusting it fully. Nobody asked me to do this. I picked it because it was a real, recurring cost, not because it was the biggest problem on the team.

Structured elaboration

Identifying the pain point. I look for tasks that are manual, repeated on a predictable cadence, and error-prone specifically because a human is doing them by hand, not because the underlying logic is hard. It doesn't need to be the team's biggest problem, just a genuine recurring cost that everyone had quietly accepted as normal.

Tools and approach. I reach for the smallest tool that actually solves the problem, usually a script using libraries already in use on the team, rather than standing up new infrastructure to justify the effort. The goal is removing the manual step, not building something impressive.

Validating it. I run the automation in parallel with the existing manual process for a defined period before fully trusting it, comparing outputs so any discrepancy surfaces while a human is still double-checking, not after the manual process has already been retired.

Rolling it out. I use it myself first, then share it with teammates doing the same manual task, with just enough documentation that someone else can run it without needing me in the room.

Measuring impact. I look at three things: time saved on the recurring task, errors caught that the manual process had been missing, and incidents avoided because a mistake never got the chance to reach anyone downstream.

Worked example

A weekly reconciliation between two data sources took about three hours every Monday and had a history of small mismatches slipping through unnoticed. Nobody had assigned fixing this to anyone, it was just accepted as the cost of Monday mornings.

I identified the pain point by noticing how much of that three hours was pure manual comparison rather than actual judgment calls. I wrote a script that pulled both sources and automatically flagged any mismatch above a small threshold, using tools the team already had access to rather than introducing anything new. To validate it, I ran the script alongside the manual process for three weeks, comparing its output against what the manual review caught each week. During that period, it caught every discrepancy the manual process had also caught, plus one that the manual process had missed entirely that week. I rolled it out by using it myself first, then sharing it with the two teammates who did the same reconciliation on other teams, along with a short readme covering how to run it and what the flag threshold meant.

Measurable impact: the task went from about three hours to roughly fifteen minutes of review, a time reduction of about 92% (165 divided by 180 minutes). Over the following quarter, the script caught four discrepancies that would previously have gone unnoticed until a downstream report was already built on the wrong numbers.

Trade-offs and pitfalls

Building something clever enough that only I understand or can maintain it creates a single point of failure, so I kept the logic simple enough for a teammate to read and adjust without me. Skipping validation because the automation "obviously" works is a common shortcut, and it's exactly how a subtle bug in the automation itself slips past unnoticed. It's also worth asking whether the manual process should even exist in its current form before automating it exactly as-is, rather than automating a bad process faster. And a real risk worth watching for: an automation's output can differ subtly in format from what the manual process produced, and if something downstream was quietly parsing the old manual report's exact shape, that consumer can break the moment the automation changes it, even though the automation's own logic is completely correct. Checking who else consumes the output, not just verifying your own correctness, is what catches that before it ships.

Direct answer

Every ratio in this snapshot points the same direction: a small number of STRAGGLER tasks are handling a disproportionate share of the data, the classic signature of DATA SKEW on the grouping/join key, not a hardware or general-tuning problem. The max task duration is 35x the median (420s vs 12s), max shuffle-read bytes per task is roughly 410x the median (4 GB vs 10 MB), and 1.2 billion max spilled records against a MEDIAN of ZERO spilled records means most tasks never spill at all while at least one task overflows its memory budget entirely; concretely, salt the skewed key (or enable Adaptive Query Execution's automatic skew-join handling) and re-measure the SAME ratios to confirm they collapse toward 1x.

Structured elaboration

Reading the ratios as a diagnostic, not just numbers. $420 / 12 = 35$: one or a few tasks take 35 times longer than a typical task in the SAME stage, meaning the stage's total wall-clock time is dominated by waiting for those few stragglers while the other 1023 tasks likely finish and sit idle. $4096 \text{ MB} / 10 \text{ MB} \approx 409.6$: the straggler tasks are not just SLOWER, they are processing roughly 410x more shuffle data than a typical task, which is the DIRECT cause of both the duration skew and (given the memory needed to hold that much shuffled data) the spill. A median of 0 spilled records with a max of 1.2 BILLION confirms this is not a general memory-undersizing problem (if it were, MANY tasks would spill, not just the outlier), it is specifically the few oversized tasks running out of their allotted execution memory and spilling to disk.

Why this points to a hot key, not hardware. If this were a genuine hardware/node issue (a slow disk, a failing node), the affected tasks would be somewhat RANDOMLY distributed across the cluster and would not correlate cleanly with shuffle-read-byte volume; here, the tasks with the LONGEST duration are (by strong implication, since they are the same small outlier set) also the tasks with the LARGEST shuffle-read volume, which is the signature of a small number of KEYS receiving a disproportionate share of the data (one or a few hot keys), not a hardware problem. A hardware check (confirming no specific NODE correlates with the slow tasks, via the Spark UI's per-task host/executor breakdown) is still worth doing to RULE OUT that alternative explanation before committing fully to the skew diagnosis, but the DATA pattern here already strongly favors skew.

Fetch wait time as corroborating evidence. Average fetch wait 200ms with a max of 20s ($20{,}000 / 200 = 100$x) shows some tasks waiting far longer than typical to FETCH their shuffle input, consistent with those tasks' upstream (map-side) partners having produced an unusually large amount of shuffle output for them to fetch, the same underlying hot-key pattern viewed from the READ side rather than the processing-time side.

Worked example

Concrete fixes, in order of how directly they target the diagnosed cause:

1. Enable/confirm Adaptive Query Execution's skew-join optimization (spark.sql.adaptive.skewJoin.enabled=true, on by default in current Spark versions): AQE detects a partition significantly larger than its peers at RUNTIME (using actual observed shuffle statistics, not a plan-time estimate) and automatically splits it into smaller sub-partitions, joined independently; this is the lowest-effort fix and should be CONFIRMED enabled and actually triggering (visible in the Spark UI's stage detail as split sub-tasks for the affected partition) before reaching for a manual fix.
2. Salt the skewed key (the mechanism in depth) if AQE's automatic handling is insufficient or unavailable for this specific operation shape: append a random suffix to the hot key on one side, explode the OTHER side to match every possible suffix, spreading what was one enormous partition across many smaller ones.
3. Isolate and handle the hot key(s) separately (a two-pass approach: filter out the known hot key(s), process them with a dedicated, differently-tuned path such as a broadcast join if the hot key's OTHER side is small, then union the results back together), appropriate when a SMALL, identifiable number of keys account for the skew (worth confirming via a quick groupBy(key).count().orderBy(desc("count")) check before committing to this more invasive restructure).
4. Increase spark.sql.shuffle.partitions as a smaller, complementary tuning lever (spreading the NON-skewed data more finely does not fix the skewed partition itself, but can reduce contention among the OTHER 1023 mostly-fine tasks), worth trying but not a substitute for directly addressing the skewed key.

Validating the fix. Re-run the SAME job and pull the identical metrics: confirm max task duration drops toward the median (the 35x ratio collapsing toward roughly 1-2x), confirm max shuffle-read bytes per task drops toward the 10 MB median, and confirm spilled records drop to near-zero across ALL tasks, not just a lower max; a fix that reduces the MAX but leaves a NEW, still-large outlier (a second hot key the fix did not address) means the diagnosis was only partially correct and needs another iteration.

Trade-offs and pitfalls

• Common mistake: jumping straight to "add more executor memory" in response to the spill, without checking whether the spill is CONCENTRATED in a few tasks (skew) or spread broadly (genuine undersizing); more memory helps a genuinely undersized job but does little for concentrated skew, since the few oversized tasks would need a disproportionately large memory bump to avoid spilling, likely reintroducing the same problem at a larger data scale later.
• Common mistake: treating "increase shuffle partitions" as a universal fix; more partitions helps spread EVENLY-distributed data more finely, but a single very large key's data all still lands in ONE partition regardless of the total partition count, since partition assignment is a function of the key's hash, not a count you can tune away.
• Ruling out hardware BEFORE committing fully to a skew fix is worth the small extra diagnostic step: correlating straggler tasks with their host/executor (available in the Spark UI's per-task detail) confirms the pattern is data-driven, not concentrated on one physical node, avoiding wasted effort implementing a skew fix for what turns out to be a hardware problem.
• Validating with the SAME metrics before and after is what turns "this ran faster" (an environment-dependent, less reliable signal) into a genuine confirmation that the specific diagnosed cause (skew) was actually fixed, not just that something changed.

Direct answer. A hot-partition runbook needs three phases in order: detect it fast with a per-partition metric (not an aggregate one), apply an immediate mitigation that does not require a cluster-wide restart, and only then execute a medium-term structural fix, verifying at each step that the fix actually reduced the imbalance rather than just moving it.

Structured elaboration.

1. Detection. Alert on per-partition throughput and produce-latency variance, not on cluster averages -- a hot partition can be invisible in an aggregate dashboard while one partition runs at 10x the others. A useful trigger: max-partition-throughput / median-partition-throughput exceeding a threshold (e.g., 3x) sustained for more than a few minutes.
2. Immediate mitigation (minutes, not hours). In rough order of increasing intrusiveness: (a) rate-limit the specific tenant/key identified as the source, which buys time without touching the topology; (b) redirect a portion of that tenant's writes to a dedicated overflow topic/partition if the client can be made key-aware; (c) as a last resort, spin up additional replicas for the affected topic/partition. Be precise about what this does and does not fix: Kafka producers always write to the partition LEADER regardless of replica count, so adding replicas does not by itself relieve write load -- it only helps if the cluster has follower-fetching enabled (consumers configured with a rack-aware client.rack), in which case READ traffic can be redirected to the new follower replicas instead of the leader, freeing up leader-broker capacity indirectly. Treat it as a read-side and availability lever, not a produce-side fix. None of these require a full cluster restart.
3. Medium-term fix (hours to days). Salting the hot key (appending a bounded random or round-robin suffix to spread it across N sub-keys) for cases where strict per-key ordering is not required, or standing up a tenant-dedicated topic for consistently high-volume tenants so their traffic never competes with the shared topic's partitioning scheme again.
4. Verification. Confirm resolution by checking the SAME per-partition metric used for detection returns to a healthy ratio, and separately confirm no consumer-side regression (e.g., a tenant that was salted now requires fan-in logic downstream to reconstruct order-sensitive views, if any existed).

Worked example. Detection shows partition 7 sustaining 4x the throughput of the topic's median partition, traced to a single tenant whose key hashes to that partition. Immediate mitigation: rate-limit that tenant's producer client to the median partition's typical throughput, restoring headroom on partition 7 within minutes and preventing cascading consumer lag. Medium-term: since the tenant does not require cross-event ordering guarantees beyond their own individual writes, salt their key with a suffix drawn from a small fixed set (e.g., 8 buckets), spreading what was one hot partition's worth of traffic across 8 partitions -- verified by confirming the tenant's traffic now shows up roughly evenly across those 8 partitions in the per-partition throughput dashboard, and that overall cluster balance (not just this tenant's own 8 partitions) returns to a healthy max/median ratio.

Trade-offs & pitfalls. Salting changes the partition assignment for future messages, which can break any downstream logic implicitly relying on 'same key always maps to same partition' (e.g., a consumer using partition assignment as a poor-man's routing table) -- audit downstream consumers before salting, not after. Rate-limiting a tenant is a real business trade-off (their events queue or get dropped depending on policy); communicate it as a temporary, bounded action with a defined resolution target rather than a silent, indefinite throttle.

Sharding

Split the keyspace across multiple cache nodes using consistent hashing (a hashing scheme where adding or removing a node only remaps a small fraction of keys, instead of a naive hash(key) % N where every node change reshuffles almost everything). Redis Cluster implements this with 16,384 fixed hash slots distributed across primaries; memcached deployments typically rely on a client-side or proxy-side consistent-hashing ring since memcached itself has no built-in clustering.

Rebalancing

When you add a shard, you migrate a subset of slots (or the ring's affected key range) to the new node while it stays live: Redis Cluster does this with MIGRATE, moving keys slot-by-slot so the cluster keeps serving traffic during the move, just with a brief per-key redirect (ASK/MOVED responses) while a slot is mid-migration. The key discipline is doing this gradually and monitoring latency during the migration window, since bulk key migration itself consumes network and CPU that competes with live traffic.

High availability

Each shard needs at least one replica so a primary failure does not lose that slice of the keyspace. Automatic failover (Redis Sentinel, a separate watcher process that monitors Redis nodes and triggers failover, or Cluster's built-in failure detection and replica promotion) needs a quorum (a required majority of nodes that must agree before an action like a failover is allowed to happen) of nodes to agree a primary is actually down, not just slow, before promoting a replica, to avoid a false failover from a transient network blip.

Client-side topology awareness

Two real options:

• Smart client: the client library holds the current slot-to-node mapping and routes requests directly, refreshing the map when it gets a redirect response. Lower latency (no extra hop) but every client needs to correctly implement cluster-aware routing and handle mapping changes gracefully.
• Proxy layer (Envoy, Twemproxy, or a managed cluster's built-in proxy endpoint): clients talk to one stable endpoint, and the proxy does the routing. Simpler clients, one extra network hop, and the proxy itself becomes a component you need to scale and keep highly available.

For a growing system, I would lean toward a smart client if the client library is mature and well-tested (most major languages have one for Redis Cluster), since it avoids adding a proxy tier as another thing that can become a bottleneck or single point of failure.

Keeping latency low as you scale out

• Avoid hot shards: a poor hash choice or a genuinely hot key (one entity getting disproportionate traffic) can overload one shard while others sit idle; watch per-shard CPU and ops/sec, not just the cluster aggregate.
• Keep replica reads local to the requesting service's region if you are spanning multiple regions, since a cross-region read adds real round-trip latency that no amount of sharding fixes.
• Pipeline and batch requests where the client supports it, since sharding adds more network hops in aggregate (one per shard touched by a multi-key operation) and pipelining amortizes that cost.

Split the model into a fast, self-service default path for low-risk, common-case access and a deliberately slower, approval-gated path reserved for the specific actions that carry real risk, bulk export, raw restricted-tier access, and cross-boundary data movement, rather than applying uniform friction everywhere or uniform speed everywhere.

Compliance obligations as concrete controls

• Data classification: every dataset tagged by sensitivity tier (public through restricted), which drives every other control below rather than each control being independently decided per dataset.
• Masking: confidential and restricted fields masked by default in any query or export; unmasking requires an explicit, logged action.
• Role-based access control (RBAC): roles scoped to the minimum access needed for a job function, avoiding broad "analyst" roles that quietly accumulate access to everything over time.
• Audit trails: every access, mask/unmask event, and export logged with who, what, when, and the stated purpose.
• Automated PII (personally identifiable information) discovery: a scheduled scan across new and existing datasets that flags PII-shaped fields that weren't already tagged, since manual tagging alone reliably misses fields added after the initial classification pass.

Automated PII discovery linked to consent

Automated discovery alone only tells you a field looks like PII; it doesn't tell you whether that data can legally be used for a given purpose. Linking discovery output to the consent-management record closes that gap: when the discovery scan flags a new PII field, the pipeline automatically checks whether the customers represented in that dataset have a consent record covering the purpose the field is being used for, and if not, the field is auto-masked pending a compliance review rather than silently becoming usable the moment it's classified. This turns "we found PII" into "we found PII, and here's whether it's currently authorized for use," which is the actionable version of the discovery signal.

Metric-definition registry as a concrete control

A metric-definition registry (a single governed source of truth for how a business metric like "active customer" or "churned account" is defined and calculated) is a control worth weighing alongside the access controls above, because a self-service analytics environment with strong access control but no metric registry still lets two analysts compute "revenue" two different ways from the same underlying restricted data, which is a data-integrity risk that access control alone doesn't address. Requiring metrics used in anything shared beyond the requester (a dashboard, a report to leadership) to be pulled from the registry, rather than computed ad hoc from masked or raw tables, is a low-friction control that catches a different failure mode than the security controls above: not unauthorized access, but authorized-but-wrong use.

Where to allow just-in-time approvals or self-service

• Self-service, no approval needed: querying already-masked, already-classified data through governed views; pulling metrics from the registry; standard-sized exports of non-restricted data.
• Just-in-time (JIT) approval, fast turnaround: a specific, time-boxed unmask request for a named dataset with a stated reason, approved by the data owner, expiring automatically.

Where to insist on friction

• Bulk export of restricted-tier data: requires a second approver beyond the data owner (for example a compliance reviewer), regardless of who's requesting it, because bulk exfiltration risk doesn't correlate with seniority or trust level, it correlates with volume.
• New consumption of PII flagged by automated discovery: blocked (auto-masked) until a compliance review confirms consent coverage, not granted by default while review is pending.
• Cross-boundary movement (for example moving a restricted dataset into a new tool or a new team's environment for the first time): requires governance sign-off, since this is exactly the kind of change that quietly expands the actual risk surface without anyone deciding to expand it on purpose.

Defending the split to a skeptical stakeholder on either side

To a security-focused stakeholder who wants more friction everywhere: the actual risk is concentrated in bulk volume and novel data movement, not in ordinary masked queries; adding approval gates to the 95% of access that's low-risk doesn't reduce the real risk, it just trains people to route around the system for routine work, which is worse for security than a fast, well-logged self-service path. Point to the audit trail: every self-service access is still logged and reviewable, friction isn't the only control, logging is a control too.

To a productivity-focused stakeholder who wants less friction everywhere: the gated actions (bulk export, new PII consumption, cross-boundary movement) are specifically the small number of actions that, if wrong, are hard or impossible to undo, once a bulk export leaves the building, revoking access to the source dataset doesn't undo the export. The friction is proportionate to the size of the mistake it prevents, not applied indiscriminately, and the self-service path covers the overwhelming majority of day-to-day work without any gate at all.

The defensible position to both is the same: friction is allocated by the actual, demonstrated risk profile of the action (volume, novelty, reversibility), not by a blanket policy in either direction, and that allocation is visible and auditable, not a judgment call made silently case by case.

Trade-offs & pitfalls

The failure mode to watch for is scope creep on the "requires friction" list: once a gate exists for bulk export, there's a temptation to extend it to smaller and smaller export sizes over time in the name of caution, which erodes the self-service path's speed until it isn't actually the fast option anymore. Re-anchoring the friction list periodically to actual incident and audit data, not to accumulated caution, keeps the split honest.

Direct answer

Don't declare the old number wrong and switch overnight. Explain the change in terms partners can verify for themselves, run both definitions side by side for a defined period so people can reconcile the gap at their own pace, and give a concrete accounting of why the numbers differ before asking anyone to adopt the new one as their working reality.

Structured elaboration

1. Find out what's actually anchored to the old number. It's rarely the number itself that people resist, it's the targets, dashboards, or comp plans built on top of it. Identify those dependencies before you talk about the redefinition in the abstract.
2. Show a concrete case where the old definition misled someone. An abstract "this is more accurate" argument doesn't land. A specific example where the old calculation gave a wrong or misleading answer does.
3. Run dual reporting, don't hard-cutover. Publish both the old and new metric side by side for a fixed window so partners can watch the two track each other (or diverge) and build intuition for the new number before they have to rely on it alone.
4. Break the gap into named components. Instead of "the number moved," account for the difference: how much of the change comes from the new inclusion/exclusion criteria, how much from a data-quality fix, how much from a genuine behavior shift. A gap people can decompose feels explainable; an unexplained gap feels arbitrary.
5. Set an explicit cutover date and update every downstream artifact by name, dashboards, target-setting docs, comp formulas, rather than assuming people will notice and adjust on their own.
6. Keep the old metric available, read-only, for a grace period after cutover instead of deleting it immediately, so people can still check their own prior conclusions against it while they adjust.

Worked example

Suppose "active users" currently counts anyone who logs in during the month. The new definition additionally requires at least one core in-product action during that session, because the team found that a meaningful share of logins were automated health-checks or bounced sessions that didn't reflect real engagement. If the old metric counted 10,000 monthly logins, and historically about 30% of logins involve no core action (a figure pulled from existing session logs, not asserted), the new definition would show roughly 10,000 x (1 - 0.30) = 7,000 active users, a drop of 3,000 driven entirely by the new inclusion criterion, not by an actual usage decline. Dual reporting both numbers for a month, with that 3,000-user gap explicitly labeled "removed for lacking a core action, not a real drop," lets a marketing partner whose Q3 target was set against the old 10,000-count number understand exactly why their dashboard changed before they have to defend it to their own leadership.

Trade-offs & pitfalls

• Pitfall: cutting over immediately without a dual-reporting window. It looks like the number was changed to hit or dodge a target, even when it wasn't.
• Pitfall: mandating adoption from authority ("this is the new source of truth, use it") without walking anyone through the why. Technically correct, but it burns trust and invites people to quietly keep using their own old tracking.
• Pitfall: deleting the old metric immediately, which strands anyone mid-adjustment and turns a change-management problem into an access problem.
• Senior differentiator: treating a metric redefinition as a change-management effort you own end to end (explanation, parallel run, decomposition, migration of dependents), not just a technical correction you announce and move on from.

Direct answer. Confirm duplication is the cause (rather than a logic bug) by comparing the row count immediately after the join to the row count you'd expect from the smaller side alone, and by checking, on a small sample, whether specific keys legitimately have multiple matches on both sides; fix it either by aggregating one side down to uniqueness before the join, or by deduplicating the joined result afterward in a way that doesn't silently discard rows you actually need.

Structured elaboration. An unanticipated many-to-many relationship means a join key you expected to be unique on at least one side actually has multiple matching rows on BOTH sides for some values, which multiplies rather than merely combines: two matching rows on each side for the same key produce four joined rows, not two, which then inflates any downstream SUM or COUNT computed from that joined result. To confirm this is the cause, pick a specific key value, count its rows on each side of the join independently, and multiply those counts together; if that product matches the number of joined rows you're seeing for that key, you've confirmed genuine multiplicative duplication rather than, say, a join condition that's simply too loose.

Two realistic fixes, and they aren't interchangeable: pre-aggregate one side down to one row per join key BEFORE the join (appropriate when you only actually need a single value per key from that side, like a "most recent" or "total" per key); or deduplicate the JOINED result afterward using a window function or an explicit grouping, which is appropriate when you genuinely need attributes from multiple matching rows on the many-side and the "duplication" is actually correct given the relationship, just not what a naive downstream SUM assumed.

Worked example. A product joined to promotions where a product can have multiple active promotions and a promotion can apply to multiple products is a genuine many-to-many; a report computing "total sales per product" that naively joins in promotions and then sums a sales column will multiply each product's real sales figure by however many active promotions it happens to have, a bug that's invisible on products with exactly one promotion and only becomes obvious (and often only gets NOTICED) on products with several.

Trade-offs and pitfalls. The riskiest version of this bug is exactly the one described in the worked example: it's silently correct for the common case (one match per key) and silently wrong only for the less common case (multiple matches), which means it can ship, look fine in testing, and only surface as a real discrepancy once someone happens to look at a key with genuine multiplicity, often much later and much harder to trace back to its root cause.

Complexity

Detecting this costs a handful of targeted counting queries against a specific suspect key value, not a full reprocessing of the dataset, so confirmation is cheap even though the underlying bug can be expensive to have shipped.

Edge cases

A key with exactly one match on both sides produces exactly one joined row and looks completely correct, which is precisely why this bug tends to survive testing against a small, low-multiplicity sample and only appears once the real data includes genuine multi-match keys.

Approach

This is a simplified t-digest: instead of storing every raw sample, keep a bounded set of centroids, each an (mean, count) pair summarizing a cluster of nearby values. New values enter as their own singleton centroid; once the centroid list grows past a threshold, a compress step sorts by mean and greedily merges neighbors until each merged centroid holds roughly total_count / max_centroids samples, capping memory at O(max_centroids) = O(1/epsilon) regardless of how many values have streamed through.

class StreamingQuantileSketch:
    def __init__(self, epsilon=0.01):
        self.epsilon = epsilon
        self.max_centroids = max(20, int(2 / epsilon))
        self.centroids = []  # sorted list of [mean, count]
        self.total_count = 0

    def update(self, value):
        self.centroids.append([float(value), 1])
        self.total_count += 1
        if len(self.centroids) > self.max_centroids * 2:
            self._compress()

    def _compress(self):
        self.centroids.sort(key=lambda c: c[0])
        merged = []
        target = max(1, self.total_count // self.max_centroids)
        for mean, count in self.centroids:
            if merged and merged[-1][1] < target:
                m, c = merged[-1]
                new_c = c + count
                merged[-1] = [(m * c + mean * count) / new_c, new_c]
            else:
                merged.append([mean, count])
        self.centroids = merged

    def merge(self, other):
        self.centroids.extend(other.centroids)
        self.total_count += other.total_count
        self._compress()

    def query(self, q):
        self._compress()
        target_rank = q * (self.total_count - 1)
        cum = 0.0
        for i, (mean, count) in enumerate(self.centroids):
            next_cum = cum + count
            if target_rank <= next_cum - 1 or i == len(self.centroids) - 1:
                return mean
            cum = next_cum
        return self.centroids[-1][0]

Driver, run against a right-skewed latency distribution (a fast path around 40ms plus a slow tail from 200ms to 2s), and output

import random
random.seed(7)
values = [max(1.0, random.gauss(40, 10)) for _ in range(9000)]
values += [random.uniform(200, 2000) for _ in range(1000)]

sketch = StreamingQuantileSketch(epsilon=0.01)
for v in values:
    sketch.update(v)

sorted_values = sorted(values)
for q in (0.5, 0.95, 0.99):
    approx = sketch.query(q)
    idx = min(len(sorted_values) - 1, int(round(q * (len(sorted_values) - 1))))
    exact = sorted_values[idx]
    print(f"q={q:<5} approx={approx:9.2f}ms exact={exact:9.2f}ms error={abs(approx-exact)/exact*100:5.2f}%")

print(f"centroids stored={len(sketch.centroids)} vs raw n={len(values)}")

Output:

q=0.5   approx=    41.50ms exact=    41.48ms error= 0.04%
q=0.95  approx=  1216.93ms exact=  1124.40ms error= 8.23%
q=0.99  approx=  1860.10ms exact=  1813.24ms error= 2.58%
centroids stored=134 vs raw n=10000

Mergeability across nodes

merge() just concatenates two sketches' centroid lists and re-compresses. This works because a centroid is just a (mean, count) pair with no ordering or index that depends on which node produced it. That is what lets each service instance keep its own local sketch and a central aggregator merge them into one global percentile view, instead of shipping every raw latency sample across the network. Verified by splitting the same 10,000 values across two independent sketches and merging them:

merged-from-two-nodes p99=1866.15ms single-sketch p99=1860.10ms

The merged result lands within about 0.3% of the single-sketch result, confirming the merge doesn't systematically distort the tail.

Being honest about the error

The median (p50) is accurate to 0.04% here, but p95 is off by over 8%, because the data has a sharp transition right around that quantile: a fast path clustered near 40ms and a slow tail starting at 200ms. This simplified sketch merges centroids to a roughly uniform target size everywhere, which under-resolves exactly where the distribution bends sharply. Real t-digest fixes this with a k-scale function that keeps centroids small near the tails (where q(1-q) is small) and lets them grow larger near the median; that piece is intentionally left out here to keep the merge logic legible. Production code should use a maintained t-digest or HDR Histogram library rather than this teaching version.

Tuning epsilon for accurate p95 at high volume

Smaller epsilon means more centroids (max_centroids scales as roughly 2/epsilon) and lower error, at the cost of more memory and slower merges. At high request volume with millions of samples per window, an epsilon around 0.001 to 0.01 (a few hundred to a couple thousand centroids) is a common practical range: below that, extra centroids stop measurably improving the p95/p99 estimate but do measurably increase memory across every process running the sketch.

Complexity

O(1) amortized time per update() (compression triggers periodically, not on every call), O(1/epsilon) space, O((1/epsilon) log(1/epsilon)) time per compress() from the sort.

Edge cases

Querying an empty sketch should raise rather than return a bogus value. Merging sketches with very different total_count values needs care: if the merge target size is derived only from the combined total, a low-traffic node's few centroids can get swallowed into a high-traffic node's larger ones and rounded away, so weighting by count during merge, rather than truncating naively, matters in practice.