Data Processing Algorithms Bottleneck Diagnosis

The reasoning

A shuffle assigns rows to partitions by a hash of the key, so every row for a given key lands in one partition. If one key is a third of the data, one partition is a third of the data, and the task that owns it takes proportionally longer. The stage completes when its slowest task does, so the job's runtime is set by the largest partition rather than by the total.

Confirmation is quantitative and takes a minute: compare the straggler's input bytes or record count against the median task's. A ratio of fifty or a hundred is skew and there is nothing else to consider. If the ratio is near one, the data is balanced and you have a different problem — a slow node, a spilling task, or an external call inside the task — and those need different fixes.

Then find the key: `SELECT key, count(*) ... GROUP BY key ORDER BY 2 DESC LIMIT 20` on the join or grouping column. Real data has real hot keys — a default value, a null, an internal test account, a genuinely huge customer — and which one it is decides the fix.

The fixes, by cause. Nulls and sentinel values usually should not be joined at all, so filter them first. A genuinely large key needs the work split: salt the hot key across N sub-keys and aggregate in two stages, or handle it as a separate branch — broadcast the small side for that key and union the results. Adaptive execution in modern engines does some of this automatically, which is worth knowing exists before you hand-roll it.

The reasoning

It is not a capacity problem. The work is trivial and the resources are idle; the queue's ordering is what serialises it. That distinction matters because the instinct — add more capacity — does nothing if the ordering constraint remains, which is exactly what a single-partition ordered stream enforces.

The simulation beside this makes it concrete: with one consumer the last item finishes at t=299 and forty-nine items complete after the slow one. With four consumers, none of them queue behind it. And giving the slow item its own lane finishes everything else at t=33 — the same total work, rearranged.

The fixes in order of preference. Take the slow work off the path: if it is an external call or a heavy computation, hand it to a side queue and let the fast path continue. Partition so that ordering is only required where it is genuinely needed — ordering per key, not globally, lets independent keys proceed in parallel, and is usually what the business rule actually requires. Add consumers, which helps only if the partitioning allows it. And bound the slow operation with a timeout so one pathological item cannot block indefinitely.

The one to be careful with is retries: a poison message that fails and is retried in place blocks the partition forever, which is head-of-line blocking with no upper bound. A dead-letter queue after N attempts is what converts that from an outage into a ticket.

The reasoning

A watermark asserts 'no event older than T will arrive'. To be safe across a partitioned source it must be the *minimum* over all partitions — the slowest one bounds the guarantee. If one partition stops producing, its watermark stops, the global minimum stops, and every window stays open waiting for events that will never come.

The symptom set is distinctive: throughput looks normal, no errors, no output, and memory rising steadily. It is easy to misread as a sink problem, because the visible fact is that nothing is being written.

The causes, in order of frequency. An idle partition — a Kafka partition with no traffic at 3am, or a source that is over-partitioned for its volume. A stalled consumer on one partition, so it is lagging rather than idle. A single event with a far-future timestamp, which pushes one partition's watermark ahead and is the mirror-image bug: it advances the watermark too far and everything that arrives afterwards is dropped as late.

The fixes: configure idleness detection so a partition with no data is excluded from the minimum after a timeout, which is the direct answer and is a standard option in Flink and Spark. Reduce partition count so partitions are less likely to be idle. Guard against future timestamps by clamping to processing time. And monitor the watermark itself — the lag between the watermark and wall-clock time is the single most informative streaming metric, and almost nobody has it on a dashboard.

The reasoning

The arithmetic first, because it decides everything: if messages arrive at 10,000/s and you process 8,000/s, the deficit is 2,000/s and the backlog grows forever. If you can process 12,000/s, the surplus is 2,000/s and a backlog of 100 million drains in about fourteen hours. Anyone answering this without those two numbers is guessing.

Then why service rate is what it is. Per-message cost times messages, divided by parallelism — so the levers are the cost, the parallelism, or the message count. Parallelism is capped by partitions in most streaming systems: sixteen partitions means at most sixteen consumers, and adding a seventeenth does nothing. That ceiling is the most common reason 'we scaled up and nothing changed'.

Then the shape of the deficit. Is it constant, meaning permanently under-provisioned, or is it a daily peak the system never fully recovers from? A backlog that grows during the day and drains overnight is healthy; one that ratchets upward week over week is not, and the distinction is only visible on a multi-day graph.

The fixes, in the order I would try them: reduce per-message work — batch the writes, remove a synchronous call, stop doing a lookup per message; raise parallelism if the partition count allows, and repartition the topic if it does not; and shed load if the deficit is structural, by sampling or by dropping a lower-value stream. And while catching up, be careful: a consumer that suddenly processes at four times the normal rate can take down whatever it writes to, so the recovery needs its own rate limit.

The reasoning

The first check is whether the *plan* changed. Most engines will show you the plan for a past execution; comparing today's against one from six months ago usually answers the question immediately. A plan flip — broadcast to shuffle, hash join to sort-merge — is a discrete change that produces exactly this discrete slowdown.

The mechanisms behind a flip. Data grew past the broadcast threshold, so a join that avoided the shuffle now performs one. A hash table that fit in memory now spills, converting a one-pass operation into a partitioned disk-based one. Statistics went stale, so the optimizer is planning for last year's volume and choosing wrongly.

If the plan is unchanged, look at the physical layout. A table appended to nightly for a year accumulates small files, so per-file overhead grows without any single query changing. Clustering decays, so zone maps stop pruning and scans read more. Deleted rows accumulate as tombstones in a merge-on-read table, so every read reconciles more deltas — compaction debt, which grows quietly and is fixed by compaction.

The last category is contention: the same query at the same time as something new. Worth ruling out early by running it in isolation, because it is cheap to check and it changes the fix entirely. The habit that makes all of this tractable is recording plan and runtime per execution, so 'when did it change' is a query rather than an investigation.

The reasoning

The pattern is always the same — something is keyed and grows monotonically. A deduplication set holding every id ever seen. Window accumulators that cannot be released because the watermark is not advancing. Session state where sessions never expire. A memoisation cache with no eviction policy. Each is correct in a test that runs for a minute and fatal in a process that runs for a month.

Distinguish it from a leak in the ordinary sense by looking at whether the growth is *proportional to distinct keys*. A heap dump showing one enormous dictionary keyed by something business-shaped is state, not a leak, and the fix is a policy rather than a bug fix.

The fixes are all about giving state a bounded lifetime. Dedup by a bounded window rather than forever — 'no duplicates within 24 hours' is almost always what the requirement really is, and it turns an unbounded set into a rolling one. Set state TTLs on keyed state where the framework supports it. Make the watermark advance, since the streaming version of this is usually the idle-partition problem rather than the state design. And where the key space is genuinely unbounded, use a structure that is bounded by construction — a Bloom filter or a fixed-size sketch — and accept the approximation.

The operational half: alert on memory *trend* rather than on threshold breach, because the threshold alert fires at 3am when it is already too late while the trend is visible for days. And a process that is restarted periodically to control memory is not a fix — it is a rehearsal for the incident where the restart does not clear it.

The reasoning

The mechanism is amplification. During the blip, requests fail and are retried, so offered load becomes original plus retries — with three retries that is up to four times normal. The service, already struggling, now gets more traffic than when it was healthy, fails more, and generates more retries. That is the storm, and it is a positive feedback loop.

The synchronisation makes it worse. Fixed or plain exponential backoff means every caller waits the same interval and retries at the same instant, so the service is hit by a wall rather than a stream. The measured simulation beside the retry question shows five hundred retries landing in the same half-second without jitter and a third fewer with it — same number of retries, spread out.

The controls, and each fixes a different part. Exponential backoff bounds how quickly load is re-offered. Jitter decorrelates callers so the load is spread. A cap stops backoff growing to uselessness. A retry budget — no more than X% of requests may be retries — bounds amplification directly and is the control most often missing. And a circuit breaker stops calling entirely after a failure threshold, giving the dependency room to recover instead of holding it down.

The half people forget: only retry what is worth retrying. A 429 or a 503 is transient; a 400 will fail identically every time and retrying it is pure amplification with no chance of success. And a retry of a non-idempotent write can succeed after the original also succeeded, so the storm can leave duplicated data behind as well as an outage.

The reasoning

**At the source.** At-least-once delivery means a message can be redelivered — a consumer that processed and then failed before committing its offset will see it again. Fingerprint: the duplicates are identical in every column including any ingestion timestamp.

**In the write.** A task wrote half a partition, failed, and was retried; the retry wrote the whole partition on top of what was already there. Fingerprint: duplicates cluster in one partition or one time window, and the count is not double but partial. This is the non-idempotent write, and it is the most common cause.

**In the load pattern.** An append where the intent was a replace: a backfill and a scheduled run both writing the same day, or an incremental model with no unique key. Fingerprint: exactly two copies of a bounded date range.

**In the transformation.** A join to a table with more than one row per key silently multiplies rows. Fingerprint: the duplicates differ in the columns that came from the joined side, and the count matches the fan-out factor. This one is not a delivery problem at all, which is why it is worth checking the grain before checking the pipeline.

How to tell them apart quickly: group by the business key and look at what actually differs between copies. Identical rows point at delivery or an append; rows differing in one column point at a join; a bounded date range points at overlapping loads. Then fix the class — idempotent writes keyed on the partition, a unique key on the merge, or the grain of the join — rather than deduplicating downstream, which hides the cause and has to be maintained forever.

The reasoning

**Lateness.** Events arriving after their window closed are dropped by design, and 'dropped by design' produces no error. A consistent small percentage is the signature of a lateness threshold slightly too tight for the tail of the arrival distribution. The check is to measure event-time minus ingestion-time on the source and compare its 99th percentile against the allowed lateness.

**An inner join.** Every fact row whose dimension key has no match disappears. New products, a dimension that loads after the fact, a key that arrives with different whitespace or casing — all produce a small, persistent shortfall. A left join with a count of unmatched keys turns this from invisible into a number.

**A filter that means more than intended.** `WHERE status != 'cancelled'` also drops rows where status is NULL, because `NULL != 'cancelled'` is not true. Three-valued logic silently removing a slice is one of the most common causes of a small consistent gap.

**Ingestion-level skipping.** Many readers can be configured to skip malformed records — permissive mode, `on_error='continue'` — and a 2% malformed rate then becomes a 2% shortfall with a log line nobody reads. The fix is to route bad records to a quarantine table rather than dropping them, so the count is explainable.

The general prevention is reconciliation: count the source and the target for the same window and assert they agree, or agree within a stated tolerance. That converts a silent 2% into a failing check, and it is the only mechanism that catches all four causes at once.

The reasoning

**Mutable source.** The underlying table is updated in place — late-arriving corrections, a CDC feed applying updates, a dimension being overwritten — so a query over 'yesterday' reads different rows today. This is the most common cause and it is not a bug in the query. The fix is to read a snapshot: a table format with time travel, or an explicit as-of version, so the report is reproducible by construction.

**Wall-clock logic.** `WHERE created_at >= current_date - 1` means something different every day it runs. Anything reprocessing the past must derive its window from the period being processed, never from the clock — which is the same discipline as using the logical date rather than `now()`.

**Non-deterministic ordering.** `LIMIT` without a total `ORDER BY`, `ROW_NUMBER()` over an ordering with ties, `first_value` over an unordered frame, or a dedup that keeps 'any' row per key. These return a valid answer each time and not the same one, and they change when the engine changes its parallelism — which is why they often appear stable for months and then move.

**Floating-point and ordering.** Summing floats in a different order gives a slightly different result, and distributed aggregation orders differently per run. Usually a rounding-level difference, occasionally visible in a reconciliation, and a real reason to use decimal types for money.

How to tell them apart: rerun against a pinned snapshot of the inputs. If the answer is now stable, the cause was mutable inputs; if it still moves, it is in the query, and the ordering candidates are where to look. Making reports reproducible — pinned snapshot, period-derived windows, total orderings — is worth doing before anyone asks, because the first time you cannot explain a moved number is the day the data stops being trusted.

The reasoning

Merge-on-read writes changes as small delta files or delete vectors rather than rewriting the data files. Writes are therefore cheap and fast. Every reader then has to reconcile the base files with all the deltas, so read cost grows with the number of unmerged changes since the last compaction.

So the symptom — fast writes, steadily worsening reads — is not a bug, it is the trade working as designed with the compaction half missing. The debt accumulates linearly with write volume, which is why it is invisible for weeks and then obvious.

Copy-on-write is the opposite: a change rewrites the affected data files, so writes are expensive and reads see clean files with no reconciliation. The choice between them is a read-versus-write frequency question, and the honest framing is that neither avoids the work — they decide who pays and when.

The fix is scheduled compaction, and it needs to be sized against write volume rather than set once. What to watch: the number of delta files or the ratio of delta bytes to base bytes per partition, which is the direct measure of debt. Most table formats expose it. And the related trap: frequent small writes create both compaction debt and a small-files problem simultaneously, so the real fix is often to batch the writes rather than to compact harder.

The reasoning

Two facts narrow it hard. The failure is in the coordinator or driver, not in an executor, so it is work happening in one process rather than distributed. And it is seconds in, before any real stage, so it is preparation rather than processing.

The overwhelmingly common cause is a broadcast join whose small side is not small. The optimizer estimated it under the threshold, so the plan collects that side into the driver to build the broadcast — and it does that first, before the large side is touched. A stale statistic or a filter the optimizer could not see through is enough.

The other candidates with the same signature: an explicit collect of a large result into the driver; a `toPandas` or equivalent; an accumulator or a broadcast variable built from something large in user code; and a very large query plan itself, on generated SQL with thousands of unioned branches.

The fixes, in order: refresh statistics, since a wrong estimate is the root; lower the broadcast threshold, or disable the broadcast for this query with a hint, which is the immediate mitigation; give the driver more memory only if the small side is genuinely near the boundary and you want it broadcast; and if it is a collect in user code, do not — write to storage and read it back distributed. And the diagnostic worth having: check the plan for a broadcast before changing anything, because the fix differs entirely if it turns out to be a collect.

The reasoning

Topologically, output requires every ancestor to have completed. A single node stuck in a non-terminal state — waiting, queued, or simply never scheduled — blocks its entire descendant subtree. The visible symptom is at the leaves and the cause is at a root, which is why scanning the failures list finds nothing: there are no failures.

So the method is to sort by state rather than by time: list every node not in a terminal state, and take the one with the earliest start or the fewest incomplete ancestors. That is the blocker. Doing it the other way — starting from the missing output and walking up — works too and takes longer on a wide graph.

The specific mechanisms. A sensor waiting for a file that never arrived, which is the most common and looks like patience rather than failure. A dependency on a previous run that never succeeded, so the chain is serialised behind an old failure. A resource limit — a full pool or a concurrency cap — so nodes are eligible and never dispatched. And a cycle introduced by a new edge, which in most schedulers is a hard error but in a hand-rolled dependency system is a silent deadlock.

The prevention that covers all of them: alert on *absence of completion* rather than on failure. 'This node has not succeeded in N intervals' catches a stuck sensor, a blocked chain, a paused job and a resource starvation identically, where failure-based alerting catches none of them. And every wait needs a timeout, so 'waiting forever' becomes a failure that alerting can see.

The reasoning

A checkpoint persists the job's state so it can resume. Its duration is dominated by the volume written, so a rising checkpoint time is a state-size graph you already have without instrumenting anything. That makes it the earliest warning of the unbounded-state problem, well before memory pressure appears.

The causes are the familiar set: keyed state with no TTL, windows that cannot close, sessions that never expire, a growing dedup set. Plus one specific to checkpointing — state that is checkpointed in full rather than incrementally, so cost scales with total state rather than with what changed.

The consequences of ignoring it compound. Checkpoints that overlap the interval mean the job spends most of its time checkpointing. Timeouts mean failed checkpoints, and a failed checkpoint means recovery must go further back — so recovery time grows at the same rate. A job that cannot checkpoint has effectively lost its ability to recover, which is a much worse position than being slow.

The fixes: incremental checkpointing where the backend supports it, so only changed state is written; state TTLs so the volume stops growing; a larger checkpoint interval, which trades recovery time for overhead and is a legitimate knob rather than a workaround; and unaligned checkpoints where backpressure is what is stretching them. The one to reach for first is the TTL, because the others manage a symptom that will keep growing.

The reasoning

First, what happens to correctness. If the events fall outside the allowed lateness, they are dropped and the windows they belonged to are silently short — the same 2%-missing signature as before, except concentrated. If they fall inside it, the affected windows are recomputed and re-emitted, so the downstream receives updates to numbers it already published, which it must be able to handle.

Second, the state. Extending allowed lateness enough to cover a day means every window for every key is retained for a day, which multiplies state by the ratio of lateness to window size. That is a very expensive way to accommodate a rare event, and it is the trade that has to be stated: correctness for this case, paid continuously by every window.

Third, the burst. A day of events in a few minutes is a throughput spike, and the downstream sink sees a correction storm. This is where a job that was comfortably provisioned falls behind, and where retries and backpressure interact badly.

The design that handles it properly is a side path rather than a wider window. Keep allowed lateness tight enough to be cheap, route events beyond it to a late-arrivals table instead of dropping them, and reconcile in batch — recompute the affected windows daily and correct the served numbers. That keeps the streaming path cheap and fast for the 99.9% case and makes the rare case a data-correctness process rather than a capacity decision. It also means late data is *visible*, which dropping never is.

The reasoning

**Ordering.** CDC events for one row must be applied in order, and ordering is only guaranteed within a partition. If the stream is partitioned by anything other than the primary key — or repartitioned somewhere in the middle — two updates to the same row can be applied backwards, leaving the older value. The fix is to partition by primary key end to end, and to apply with a version or LSN comparison so an out-of-order event is discarded rather than applied.

**Missed events.** A consumer that skipped an offset, a connector restart that resumed from the wrong position, or a retention window that expired before the consumer caught up. The replica is then permanently missing a change, and nothing downstream can detect it.

**Deletes.** Many CDC pipelines handle inserts and updates and quietly drop deletes, or the source performs a hard delete that produces no event at all. The replica keeps rows that no longer exist — which is the same shape as the snapshot hard-delete problem.

**Schema and column scope.** An update to a column the pipeline does not carry produces an event whose payload looks unchanged for the tracked columns, and some implementations then skip it — so a version marker moves without the data. And a new column added at the source is invisible until someone notices the replica has been missing it.

The only reliable answer to all four is reconciliation rather than prevention: periodically compare source and replica — row counts per partition, and a checksum over the tracked columns — and repair what disagrees. CDC is an optimisation over full reload, and it needs the full reload as its backstop. Without a reconciliation loop, drift is undetectable by construction, which is why 'it has reported success for months' is not evidence of anything.

The reasoning

Every boundary costs a conversion. A shuffle serialises rows to bytes and back. A Python UDF in a JVM engine serialises each row out to a Python process and the result back, which is why one UDF can dominate a job that otherwise does nothing expensive. Reading a row-oriented format into a columnar engine converts on the way in.

Garbage collection is usually a symptom of the same thing: per-row object allocation. An engine operating on batches of primitives allocates almost nothing; one materialising an object per row allocates millions, and the collector then spends real CPU on them. That is the mechanism behind 'GC is 40% of the job'.

The fixes, in order of effect. Replace UDFs with native expressions wherever possible — the engine's built-in functions run inside the vectorised path and cross no boundary. Where a UDF is unavoidable, use the vectorised or arrow-based variant, which converts a batch at a time instead of a row. Reduce shuffles, since each is a full serialisation round trip. And choose an efficient serialisation format and a columnar interchange, so a crossing that must happen is as cheap as it can be.

The diagnostic that makes this concrete: compare the job's runtime with the UDF replaced by a constant. If it collapses, the cost is the boundary rather than the computation, and no amount of optimising the logic inside the UDF will matter.

The reasoning

The classic cause is a monotonically increasing partition key — an auto-increment id, or a timestamp. Every new write has a key just above the last, so every write goes to the shard owning the current range. Sixty-four shards, one of them taking all the traffic. Range-partitioned stores are especially prone to this; hash-partitioned ones are prone to the other cause, which is a single dominant value.

It differs from read skew because the key usually has a job beyond distribution. A partition key that supports efficient range scans by time is *also* the thing concentrating your writes, so 'just hash it' fixes the write and destroys the read. That tension is the actual design question.

The standard resolutions. A composite key that puts a high-cardinality component first and the time component second, so writes spread and time-ranged reads within a key still work. A bucket prefix — `hash(id) % N` prepended to the timestamp — which spreads writes across N shards and turns a range read into N parallel range reads, an acceptable trade at small N. Or accept the concentration and buffer: batch writes and flush in bulk, which does not fix distribution but reduces the number of operations the hot shard must serve.

The thing to say out loud is that this is a schema decision, not a tuning one, and it is expensive to change once data exists. Which is why it is worth getting right at design time, and why the interview question is really 'do you think about the write path when you choose a key'.

The reasoning

A backfill submits many runs at once. If they share the cluster's capacity with production and the scheduler is roughly fair, the backfill's hundred queued tasks and production's five compete on equal terms, so production gets a twentieth of the resource it usually has. It does not fail; it just misses its deadline, which is worse because nothing alerts.

The second mechanism is downstream: both are hitting the same warehouse, the same source API, the same object store. Even with perfectly separated compute, the shared dependency saturates and both slow down. That one is easy to miss because the compute metrics look fine.

The fixes are all about making the separation explicit. A resource pool per class of work, sized so that backfill can never take more than its share — the constraint has to be enforced rather than agreed. Concurrency limits on the backfill specifically, which is the cheapest version. Lower priority for backfill work so production preempts it. And chunking the backfill so it proceeds in bounded batches with checkpoints, rather than as one enormous submission.

The scheduling half worth adding: run large backfills in the window where production is quiet, and make them resumable so they can be paused when production needs the capacity. A backfill that cannot be stopped and restarted safely is one you have to either finish or abandon, which is exactly the position you do not want at 4pm.

The reasoning

First, get a per-stage breakdown. 'The pipeline is slow' almost always means one stage is slow, and the rest is noise. Total runtime tells you nothing; the distribution across stages tells you where to look, and it is available from the engine's own history without instrumenting anything.

Second, characterise that stage with one question: is the time spread evenly across its tasks or concentrated in a few? Even means it is bound by volume, and the levers are reading less — pushdown, projection, filtering earlier — or more parallelism. Concentrated means skew or head-of-line blocking, and more parallelism will not help at all. That single check splits the search space in half and takes a minute.

Third, check whether the work is real. Spilled bytes, serialisation time, garbage collection, and shuffle read time are all overhead rather than computation, and if they dominate then optimising the logic is pointless. This is where a job that is 'CPU bound' turns out to be bound by a UDF boundary or by GC.

Fourth, look outside the job. A stage waiting on an external service, an object store rate-limiting the listing, or a warehouse under load from something else entirely. The signature is a stage with low resource utilisation and high elapsed time — busy waiting rather than working.

Then, and only then, change one thing and re-measure. The discipline that matters is having a hypothesis with a number attached before touching anything: 'this stage spends 60% of its time on a shuffle of 400 GB, so filtering before the shuffle should remove most of it.' A change without a prediction is a guess, and a guess that happens to work teaches nobody anything about the next one.