The WebDataset shard convention

WebDataset is not a binary format. It is a convention for laying out files inside an ordinary POSIX tar archive, which means the data can be inspected with tar, produced by any tool that writes tar, and read without a format-specific parser. The convention is the entire specification.

The convention

Files in the tar are grouped into samples by their name up to the first dot. Everything after that dot is the field name. Files belonging to one sample must be adjacent in the archive.

images-00000.tar
  ├── 000000.jpg          sample "000000", field "jpg"
  ├── 000000.cls          sample "000000", field "cls"
  ├── 000000.json         sample "000000", field "json"
  ├── 000001.jpg          sample "000001", field "jpg"
  ├── 000001.cls
  └── 000001.json

Because the field name is the extension, the decoder is chosen per field. .jpg decodes to an image, .json to a dict, .cls to an integer, .npy to an array. Adding a modality means adding a file, not changing a schema.

import webdataset as wds

ds = (
    wds.WebDataset("shards/images-{00000..00063}.tar", shardshuffle=100)
    .shuffle(1000)             # sample buffer, spans shard boundaries
    .decode("pil")             # .jpg -> PIL, .json -> dict, .cls -> int
    .to_tuple("jpg", "cls")
)

for image, label in ds:
    ...

Why sequential access matters

A dataset stored as loose files requires one open per sample, at a random location. On a local SSD this is tolerable. On network storage, or on object storage where every read is an HTTP request, per-file latency dominates and the GPU waits.

A tar shard converts that into one sequential read of a large object. The per-sample cost becomes a memcpy rather than a request round-trip, and the storage layer sees a streaming access pattern it can read ahead on.

This is the same reasoning behind TFRecord sharding. WebDataset differs in that it requires no conversion: the bytes inside the tar are the original files, so building a shard is tar cf, and inspecting one is tar tvf.

Sharding and distributed training

Shards are the unit of parallelism, so shard count sets the ceiling on how many readers can work independently. The usual constraint is that shard count should be a multiple of world_size × num_workers, or some workers finish early and the epoch ends ragged.

• Shard size: roughly 100 MB to 1 GB. Large enough that sequential read dominates, small enough that a failed read is cheap to retry and shards distribute evenly.
• Shard count ≥ total worker count, and ideally an exact multiple. With 8 GPUs × 4 workers, use a multiple of 32 shards.
• Shuffle at two levels. Shard order across the epoch, and a sample buffer within each shard. Neither alone gives a good mix; the buffer size bounds how far a sample can move.
• Do not compress the tar itself. A .tar.gz still streams, but mid-shard resumption and range requests stop working, and CPU is spent recompressing members that are already compressed. Compress the members instead: store JPEGs, not raw pixels.

Common failure modes

• A dot in the sample name. image.001.jpg keys on image, with field 001.jpg. Sample keys must not contain dots.
• Non-adjacent members. Sorting the archive by extension rather than by name splits every sample. Sort by full path before writing.
• Missing fields. A sample lacking a requested field is dropped or errors depending on the handler; silent dropping shrinks the epoch without warning.

Since a shard is a plain tar, tar tvf shard.tar | head -20 diagnoses most of these. The WebDataset viewer does the same grouping in the browser and shows the reconstructed samples with their fields, which makes an ordering problem visible immediately.