End-to-End ML System Design Questions

What I would inspect first
I would start with per-step timing on the slow worker versus the rest of the cluster. If the code, model, and data are unchanged, a single lagging worker is usually a host or systems issue, not an ML issue.

Checks in order

• GPU utilization and memory bandwidth on the slow node
• Data loader wait time and local disk throughput
• CPU steal, thermal throttling, and noisy neighbors
• Network errors, packet drops, and collective communication logs
• Kernel and container logs for retries or hardware faults

Mitigation
If the worker is clearly abnormal, I would cordon it, move the job to a fresh node, and keep the training moving. If the slowdown comes from input starvation, I would reduce preprocessing on that host, increase local caching, or lower dataloader contention.

Worked example
If most workers take 180 ms per step but one takes 420 ms and spends 250 ms waiting on input, the bottleneck is the input path, not the model.

The goal is to isolate the bad actor quickly and avoid letting one slow node stall the whole synchronous job.

Approach
I would trace one request from ingress to response using distributed tracing. A trace is a single request’s timeline, broken into spans, which are timed steps like feature retrieval, preprocessing, model inference, and downstream calls. CPU can look normal because the bottleneck may be waiting on network, locks, queueing, or a slow dependency, not raw compute.

How I would isolate it

1. Start with one trace ID from a slow p99 request.
2. Check span timings for each stage: feature store, preprocessing, batching, model execution, postprocessing, and any outbound calls.
3. Compare slow traces to fast ones. I care about where the extra time accumulates and whether it is wait time or compute time.
4. Add or inspect metrics per stage, such as feature fetch latency, batch queue time, inference time, and downstream timeout rate.
5. If tracing is coarse, I would add sub-spans in the service and temporary structured logs with the trace ID to pinpoint the jump.

Worked example
If median requests are 80 ms and p99 is now 160 ms, I might find: feature retrieval 15 ms, preprocessing 10 ms, batch queue 55 ms, model execution 20 ms, downstream call 45 ms. That tells me the issue is batching or a dependency, not the model itself.

Next checks

• Feature retrieval slow only on cache misses, then look at cache hit rate and feature store latency.
• Batching spikes, then inspect queue depth and batch timeout settings.
• Model execution spikes, then profile the serving container.
• Downstream dependency spikes, then check retries, timeouts, and circuit breakers.

If autoscaling did not trigger, I would also verify the scaling signal. CPU-based scaling often misses latency problems, so I would prefer queue depth, request rate, or custom p95/p99 latency signals for this workload.

Core services
I would build a platform with a job API, scheduler, quota service, checkpoint store, metadata catalog, and node agents. The scheduler matches jobs to GPU pools, while the quota service enforces team budgets and priorities.

How jobs get compute
A team submits a job spec with requested GPUs, runtime, and a checkpoint policy. The scheduler grants a lease, which is a time-bounded claim on nodes. Workers heartbeat to renew the lease. On completion, failure, or preemption, the lease ends and the GPUs are released automatically.

Controls

• Per-team quotas and project budgets
• Priority classes for urgent training
• Spot first scheduling with fallback to on-demand
• Mandatory checkpoints for long jobs
• Admission control when the cluster is full

Worked example
A 64-GPU run might get 8 workers with 8 GPUs each and a 30-minute renewable lease. If a node fails, the job restarts from the last checkpoint instead of starting over.

This design keeps cost visible, failures recoverable, and shared capacity fair.

What I would check first
I would measure where time is going in one training step. The three main bottlenecks are input pipeline, compute, and communication.

Definitions

• Input pipeline: loading, decoding, and preprocessing data before it reaches the GPU.
• Compute: the actual forward and backward passes on the GPU.
• Communication: syncing gradients or parameters across devices.

Evidence to collect

• GPU utilization and memory usage
• Step time broken into data wait, compute, and sync time
• Disk and network throughput
• CPU usage on dataloader workers

Worked example
If a step takes 300 ms total and 180 ms of that is waiting for data, then the GPU is idle most of the time and the input path is the bottleneck. If compute is fast but time rises when more workers are added, communication may be the issue.

Rule of thumb
Low GPU utilization usually points to data or host issues. High GPU utilization with slow scaling usually points to synchronization or network overhead.

I would start with the timing breakdown because it tells you which team, storage, compute, or networking, needs to act first.

Ways to split the work

• Data parallelism. Each GPU holds a full copy of the model and processes different batches. It is simplest, but the model must fit on one GPU.
• Tensor parallelism. Each layer is split across GPUs. This helps when the layers themselves are too large, but it adds communication every layer.
• Pipeline parallelism. Different layers live on different GPUs. This reduces memory pressure, but can leave devices idle if the pipeline is not balanced.
• Sharded optimizers or fully sharded data parallelism. These split parameters, gradients, and optimizer state to reduce memory use while keeping the training code closer to data parallelism.

How I would choose
If the model is only slightly too big for one GPU, I would prefer sharded data parallelism because it is usually the easiest path. If individual matrix multiplies are the memory or compute problem, tensor parallelism may fit better. If the model is very deep and activations dominate, pipeline or hybrid parallelism can help.

Worked example
For an 8 GPU cluster and a model that almost fits on one device, I would first try sharding plus data parallelism before moving to a more complex hybrid design.

The tradeoff is always simplicity versus communication cost versus memory savings.