Define what a shard controls in this pipeline
A shard may be a tar archive, a columnar file, or another object containing many samples. Its size controls more than transfer length. It can be the unit assigned to a worker, cached on local NVMe, shuffled among ranks, retried after an error, validated by checksum, and replaced when data changes. Before tuning bytes, document which of those boundaries actually follow the shard in the current loader.
WebDataset describes numbered tar shards read through sequential streams and notes that multi-node fixed-epoch balancing can be difficult with iterable datasets. That makes sample count per shard as important as byte size. Two 1 GiB shards can carry very different decode time or sample counts. Record compressed bytes, uncompressed bytes, samples, and measured decode work for each candidate.
Name the costs that move in opposite directions
Very small shards increase object count, listings or manifest size, open calls, authentication work, and the chance that workers repeatedly wait for the next object. Very large shards reduce those transitions and favor sequential access, but offer fewer independent work units. They can weaken shard-level shuffling, create load imbalance, occupy cache longer, and force a retry to repeat more bytes.
Object storage also rewards concurrency within limits. Amazon S3 documents parallel requests as a way to increase read performance and publishes request-rate guidance per partitioned prefix, while noting that scaling to a new request rate can be gradual. That guidance does not establish an ideal training shard size. The loader, network path, cache, decompressor, sample sizes, and request pattern determine the result. Treat provider guidance as a boundary condition, then measure the actual pipeline.
Run a controlled size sweep with fixed samples
Build several physical layouts from the same immutable sample set, for example target shards near 64 MiB, 256 MiB, 1 GiB, and 4 GiB. Keep sample order seed, model, batch shape, worker count, node count, cache policy, augmentation, and storage location fixed. Run both cold-cache and warm-cache trials. Randomize candidate order so a transient storage or network change does not always favor one size.
Measure accepted samples per second at the trainer, GPU data-wait time, p50 and p99 object-open latency, GET or equivalent request count, bytes read, local-cache hit rate, worker idle time, and variation among ranks. Report steady state separately from startup. A layout that reaches a high peak after a long fill may still be poor for short jobs or frequently preempted workers.
| Candidate | Expected strength | Risk to measure |
|---|---|---|
| 64 MiB | Fine retries and many shuffle units | Open and request pressure |
| 256 MiB | Moderate concurrency and retry scope | Manifest and cache behavior |
| 1 GiB | Long sequential reads | Fewer assignment units |
| 4 GiB | Low transition frequency | Large retries and coarse locality |
Force errors instead of estimating retry cost
During a stable interval, inject one read failure after 10 percent, 50 percent, and 90 percent of a shard has been consumed. Use a disposable copy or loader fault hook rather than corrupting production data. Record repeated bytes, repeated samples, time until the worker produces its next accepted batch, rank stall time, and whether samples are duplicated or skipped. Repeat for each size candidate.
A streaming tar reader may need to reopen and scan from the beginning unless the application has a usable index, byte-range strategy, or sample-level resume state. Local cache can change that result. Test remote retry and cached retry separately. Recovery granularity is the real amount of work repeated by this implementation, not the nominal object size.
Worked example: throughput alone picks the wrong winner
Suppose an illustrative sweep processes the same dataset. The 256 MiB layout reaches 92,000 accepted samples per second, makes 4.1 opens per worker-minute, and repeats 150 MiB at p95 after injected failure. The 1 GiB layout reaches 96,000 samples per second, makes 1.2 opens per worker-minute, and repeats 690 MiB at p95. The 4 GiB layout reaches 96,500 samples per second but shows rank imbalance and repeats 2.8 GiB at p95. These are hypothetical results, not reported benchmarks.
Define a gate before seeing results: choose the smallest candidate within 3 percent of maximum accepted throughput, provided p99 data-wait stays below 2 percent of step time and p95 retry repeats less than 512 MiB. Maximum throughput is 96,500, so the 3 percent floor is 93,605 samples per second. The 1 GiB candidate clears throughput but fails retry scope; 256 MiB misses the throughput floor. This experiment would trigger a second sweep between 256 MiB and 1 GiB rather than crown either.
Audit shuffle quality and node locality together
Shard shuffling and sample shuffling are different layers. With too few shards per epoch, ranks can receive stable groups of related examples even when samples are shuffled inside each group. Track class, source, language, time range, or another meaningful attribute across consecutive batches and ranks. Compare the observed mixing statistic with a fully randomized reference that is feasible on a sample.
Local NVMe caching favors reuse of complete objects, but a shard larger than the available per-worker cache share can churn before reuse. Record evictions and remote bytes per accepted sample. If workers share a cache, check whether their assignment policy sends the same shard to several workers or preserves locality. More random distribution can improve mixing while increasing remote reads, so neither metric should silently dominate.
Choose with a measurable rule and preserve escape hatches
Promote a layout only when three repeated trials meet the throughput, stall, request-rate, retry-scope, and shuffle gates. Store a manifest that maps stable sample IDs to shard ID and checksum. That makes the layout reproducible and supports targeted replacement of a damaged shard. Keep conversion code and the prior manifest until at least one full training and recovery cycle completes.
The result is workload-specific. Larger samples, stronger compression, range-capable formats, a different object store, more loader workers, or a larger cache can move the optimum. Shard size also cannot fix slow decoding or an undersized network path. ClusterBid capacity matching is most useful after the experiment identifies required storage throughput, local scratch capacity, and request concurrency alongside GPU count.