Datasets

Describing the data you have, and reading it back without a pipeline. The guide covers these in Working with Datasets.

Public IO API: datasets, the in-memory shard primitive, and store options.

Configure store behavior via Pipeline.options(io_options=StoreOptions(...)).

class ArrowOptions(cpu_threads=_NotSupplied.TOKEN, io_threads=_NotSupplied.TOKEN)

Bases: object

Process-wide Arrow CPU and I/O pool sizes, shared by all Arrow readers.

Counts default to one. Explicit None leaves that pool unchanged. Configuring another store in the same process affects existing readers too.

cpu_threads: int | None = 0

io_threads: int | None = 0

classmethod from_any(obj)

Normalize Arrow pool options without importing PyArrow.

Return type:

ArrowOptions

merge(other)

Overlay supplied fields, preserving explicit None.

Return type:

ArrowOptions

class CacheOptions(enabled=_NotSupplied.TOKEN, root=_NotSupplied.TOKEN, limit_bytes=_NotSupplied.TOKEN, rg_cache_bytes=_NotSupplied.TOKEN, keep_zip=_NotSupplied.TOKEN, validate_hash=_NotSupplied.TOKEN, download_retry=_NotSupplied.TOKEN, download_timeout=_NotSupplied.TOKEN, open_retry_attempts=_NotSupplied.TOKEN, open_retry_initial_backoff=_NotSupplied.TOKEN, open_retry_max_backoff=_NotSupplied.TOKEN, min_slack_bytes=_NotSupplied.TOKEN, max_slack_bytes=_NotSupplied.TOKEN)

Bases: object

On-disk shard-cache configuration.

enabled: bool = 0

root: str | Path = 0

limit_bytes: int | None = 0

rg_cache_bytes: int | None = 0

keep_zip: bool = 0

validate_hash: str | None = 0

download_retry: int = 0

download_timeout: float = 0

open_retry_attempts: int = 0

open_retry_initial_backoff: float = 0

open_retry_max_backoff: float = 0

min_slack_bytes: int = 0

max_slack_bytes: int = 0

classmethod from_any(obj)

Return type:

CacheOptions

merge(other)

Overlay fields supplied by other onto self.

Return type:

CacheOptions

class Dataset(name, backend, path=None, _ids=None, _counts=None)

Bases: object

Descriptor for a dataset with index and backend info.

Instances are created via from_path (file-backed) or from_dict (in-memory/testing). They always provide:

• name: a user-facing identifier used in mixtures

• backend: opaque metadata that lets FetchOp build a reader later

• path: original filesystem path if file-backed, otherwise None

File-backed datasets also carry a few-KB handle to the node-local shard catalog (set by from_path); this is what travels in ctx, not shard_meta.

Shard counts are not stored as a mapping; read them via ids() / counts() / total() / max_count() / shard_count().

Backend kinds used by the internal store builder:

• litdata/mds/jsonl/parquet/vortex: kind and path (no per-shard shards graph — that lives in the catalog now)

• inmem: kind and shards (dict[int, InMemoryShard])

Note: this class does not expose any method to fetch rows; IO is delegated to an internal shard store owned by the FetchOp.

name: str

backend: Mapping[str, object]

path: str | None = None

ids()

Return the sorted shard ids as an int64 array.

Return type:

ndarray

counts()

Return per-shard sample counts (int64), aligned with ids().

Return type:

ndarray

raw_bytes()

Return per-shard byte sizes (int64), aligned with ids().

File-backed datasets read the catalog’s on-disk shard sizes; in-memory shards size their resident payloads lazily (InMemoryShard.raw_bytes).

Return type:

ndarray

total()

Total sample count across all shards.

Return type:

int

max_count()

Largest single-shard sample count (0 when empty).

Return type:

int

shard_count()

Number of shards.

Return type:

int

classmethod from_path(name, path, *, fmt=None)

Construct a file-backed dataset descriptor.

Performs a count-only discovery: it obtains shard_id + num_rows as small numpy arrays (the work source’s input) without materializing the per-shard shard_meta graph. The full columnar catalog is built once per node later, by the Engine’s finalize() (or lazily by the store builder for Engine-less use).

Parameters:

• name (str) – Logical dataset name used in mixtures and debugging.

• path (str) – Filesystem directory containing the dataset.

• fmt (str | None) – Optional explicit format. When None, auto-detects.

Return type:

Dataset

Supported formats:

• litdata directories containing index.json structured with config and chunks

• mds directories containing index.json structured with shards

• jsonl directories where *.jsonl files act as shards, with an optional Zephon index.json for fast discovery

Special URI schemes:

• hf://org/name[@rev]/[config/]split is served through the HuggingFace backend: the split’s uploaded files when their format is readable, else HuggingFace’s Parquet conversion when it is complete and built from the requested commit. fmt picks the source in that format; appending ~original or ~parquet to the revision (hf://org/name@~parquet/split) forces one. A partial conversion is used only with ZEPHON_HF_ALLOW_PARTIAL=1. Uploaded files are read as stored: the dataset card’s reader options and features casting are not applied. The dataset’s path becomes a URI pinned to the resolved commit, source and config.

Returns:

a descriptor populated with shard counts and a catalog handle for later IO.

Return type:

Dataset

Raises:

• FileNotFoundError – if path does not exist.

• ValueError – if path is not a directory or format unsupported.

classmethod from_dict(name, shards)

Construct an in-memory dataset descriptor.

Return type:

Dataset

class InMemoryShard(rows)

Bases: object

List-backed shard useful for tests and quickstarts.

property raw_bytes: int

Estimated decoded payload bytes for this shard, sampled and cached.

close()

getsamples(indices)

Return type:

list[dict[str, object]]

class ParquetOptions(use_threads=_NotSupplied.TOKEN, rg_cache=<factory>)

Bases: object

Parquet read behavior and the node-shared decoded row-group cache.

use_threads enables parallel column decoding through Arrow’s CPU pool. False keeps each read on its calling thread; separate fetches can still decode concurrently. Pool sizes are configured separately in ArrowOptions.

use_threads: bool = 0

rg_cache: ParquetRGCacheOptions

classmethod from_any(obj)

Normalize Parquet read and cache options.

Return type:

ParquetOptions

merge(other)

Overlay supplied read options and merge nested cache fields.

Return type:

ParquetOptions

class ParquetRGCacheOptions(enabled=_NotSupplied.TOKEN, root=_NotSupplied.TOKEN, limit_bytes=_NotSupplied.TOKEN, min_free_bytes=_NotSupplied.TOKEN)

Bases: object

Node-shared decoded Parquet row-group cache configuration.

enabled=None enables the cache with the shard cache or an explicit root. An omitted root uses the shard cache’s reserved .parquet-rg-cache child. An omitted limit uses the deprecated shard option when present, otherwise the larger of 4 GiB and 10% of the shard cache limit. These defaults resolve when the store is built so option merges retain the distinction between omitted and explicit values. Explicit None resets a previously supplied field to its automatic default.

enabled: bool | None = 0

root: str | Path | None = 0

limit_bytes: int | None = 0

min_free_bytes: int = 0

classmethod from_any(obj)

Normalize a decoded row-group cache configuration.

Return type:

ParquetRGCacheOptions

merge(other)

Overlay fields supplied by other onto self.

Return type:

ParquetRGCacheOptions

class StoreOptions(cache=<factory>, arrow=<factory>, parquet=<factory>, vortex=<factory>, metadata_cache_entries=_NotSupplied.TOKEN)

Bases: object

Top-level IO store options passed to FetchOp.

metadata_cache_entries bounds retained footers per format and fetch replica, independently of the on-disk shard cache; zero disables retention. Parquet footer sizes grow with the number of columns and row groups.

cache: CacheOptions

arrow: ArrowOptions

parquet: ParquetOptions

vortex: VortexOptions

metadata_cache_entries: int = 0

resolved_parquet_rg_cache()

Resolve decoded-cache defaults against the shard-cache options.

Return type:

ParquetRGCacheOptions

classmethod from_any(obj)

Return type:

StoreOptions

merge(other)

Return a new options object with other overriding self.

Return type:

StoreOptions

class VortexOptions(segment_cache_bytes=_NotSupplied.TOKEN, worker_threads=_NotSupplied.TOKEN)

Bases: object

Vortex read-cache and process-wide worker-pool configuration.

segment_cache_bytes sets the encoded-segment cache’s target size per fetch replica. Zero disables it. The target is not a strict memory bound. worker_threads defaults to one background worker; calling fetch threads can also execute Vortex work. Zero disables background workers, and explicit None leaves the pool unchanged. An explicit count overrides VORTEX_MAX_THREADS; an omitted count respects that environment variable. Stores in one process share the pool, so the last applied count wins.

segment_cache_bytes: int = 0

worker_threads: int | None = 0

classmethod from_any(obj)

Normalize Vortex options, accepting byte sizes such as '64mb'.

Return type:

VortexOptions

merge(other)

Overlay fields supplied by other onto self.

Return type:

VortexOptions

Inspecting a dataset

class DatasetInspector(dataset, *, io_options=None)

Bases: object

Inspect raw payloads from one dataset without a work source or pipeline.

This utility is intended for interactive inspection and dataset validation. It does not choose sample order, add record metadata, prefetch, batch, or run workers. Training pipelines should continue to use zephon.Pipeline.

Use inspectors as context managers so opened shard resources are closed promptly. An inspector is single-process and not thread-safe.

read(shard_id, sample_index)

Return one raw payload identified by its shard and local row index.

Parameters:

• shard_id (int) – Dataset-local shard identifier, as returned by Dataset.ids().

• sample_index (int) – Zero-based row index inside that shard.

Returns:

The payload stored by the underlying format. Payloads are normally mappings, but binary formats may return arrays or other supported payload values.

Return type:

bytes | memoryview | str | int | float | complex | Any | None | list[SamplePayload] | dict[Any, SamplePayload]

read_many(shard_id, sample_indices)

Return raw payloads for local row indices from one shard.

The returned list preserves the order of sample_indices. Supplying an empty list performs no I/O and returns an empty list.

Return type:

list[bytes | memoryview | str | int | float | complex | Any | None | list[bytes | memoryview | str | int | float | complex | Any | None | list[SamplePayload] | dict[Any, SamplePayload]] | dict[Any, bytes | memoryview | str | int | float | complex | Any | None | list[SamplePayload] | dict[Any, SamplePayload]]]

close()

Close every shard opened by this inspector.

Calling close more than once is safe. A closed inspector cannot be used for further reads.