Pre-release

This release is a pre-release and may not be stable for production use.

Zephyr

Simple data processing library for Marin pipelines. Build lazy dataset pipelines that run on Iris jobs or a local backend.

Quick Start

from zephyr import Dataset, ZephyrContext, load_jsonl

# Read, transform, write
ctx = ZephyrContext(max_workers=100)
pipeline = (
    Dataset.from_files("gs://input/", "**/*.jsonl.gz")
    .flat_map(load_jsonl)
    .filter(lambda x: x["score"] > 0.5)
    .map(lambda x: transform_record(x))
    .write_jsonl("gs://output/data-{shard:05d}-of-{total:05d}.jsonl.gz")
)
ctx.execute(pipeline)

Key Patterns

Dataset Creation:

Dataset.from_files(path, pattern) - glob files
Dataset.from_list(items) - explicit list

Loading Files

.load_{file,parquet,jsonl,vortex} - load rows from a file

Transformations:

.map(fn) - transform each item
.flat_map(fn) - expand items (e.g., load_jsonl)
.filter(fn) - filter items by function or expression
.select(columna, columnb) - select out the given columns
.window(n) - group into batches
.reshard(n) - redistribute across n shards

Output:

.write_jsonl(pattern) - write JSONL (gzip if .gz)
.write_parquet(pattern, schema) - write to a Parquet file
.write_vortex(pattern) - write to a Vortex file

Execution (ZephyrContext):

ZephyrContext(max_workers=N) — auto-detects the backend (Iris inside an Iris job, local otherwise) via fray.current_client()
ZephyrContext(client=LocalClient()) — explicit local backend (testing)
ctx.execute(pipeline) — runs the pipeline; returns a ZephyrExecutionResult(results, counters)

Read-only memory stores

ZephyrContext.load_memory_store() loads an existing partitioned dataset into the workers of an entered context. Every worker starts an empty multi-table service; pipelines that do not load a table perform no source reads. The table handle is picklable, so later pipelines and child jobs can use get() or order-preserving get_many() lookups without copying the table into every task.

from fray.types import ResourceConfig


def document_partition(key: tuple[int, str]) -> int:
    file_index, _ = key
    return file_index


documents = Dataset.from_files("s3://bucket/documents/*.parquet").load_parquet().map(
    lambda row: ((row["file_index"], row["id"]), row["text"])
)

with ZephyrContext(
    max_workers=16,
    resources=ResourceConfig(cpu=2, ram="8g"),
) as ctx:
    document_store = ctx.load_memory_store(
        documents,
        name="documents",
        hash_key=document_partition,
        recovery_timeout=900,
    )
    result = ctx.execute(Dataset.from_list(document_keys).map(document_store.get))
    document_store.destroy()

For P source shards, every key must already satisfy hash_key(key) % P == source_shard_index. Construction checks every row and does not insert a shuffle. Readers and shard-local maps can load directly. Persist and reload the output of a shuffle, join, reshard, reduce, or write before constructing a store.

Keys must be hashable and unique. Keys, values, and the hash function must be picklable for remote calls; Python's salted hash() is not stable enough to serve as the partition function for string or byte keys. Actors retain the loaded Python objects directly, and store.stats() reports item counts and load time. Invalid input fails the load call without consuming actor restart retries. Multiple tables can share the worker process. Zephyr does not reserve, limit, or evict table memory; size the context's worker RAM for the combined tables and pipeline workload.

Iris reconstructs a preempted worker at the same endpoint. The first table lookup on that replacement reloads its immutable source shards, and all worker responses and reconstruction share one recovery_timeout deadline. Destroying one table leaves other tables active. Exiting the creating context stops the worker pool and invalidates its table handles.

Real Usage

Wikipedia Processing:

from zephyr import Dataset, ZephyrContext, load_jsonl

ctx = ZephyrContext(max_workers=100)
pipeline = (
    Dataset.from_list(files)
    .load_jsonl()
    .map(lambda row: process_record(row, config))
    .filter(lambda x: x is not None)
    .write_jsonl(f"{output}/data-{{shard:05d}}-of-{{total:05d}}.jsonl.gz")
)
ctx.execute(pipeline)

Dataset Sampling:

from zephyr import Dataset, ZephyrContext

ctx = ZephyrContext(max_workers=1000)
pipeline = (
    Dataset.from_files(input_path, "**/*.jsonl.gz")
    .map(lambda path: sample_file(path, weights))
    .write_jsonl(f"{output}/sampled-{{shard:05d}}.jsonl.gz")
)
ctx.execute(pipeline)

Parallel Downloads:

from zephyr import Dataset, ZephyrContext

tasks = [(config, fs, src, dst) for src, dst in file_pairs]
ctx = ZephyrContext(max_workers=32)
pipeline = Dataset.from_list(tasks).map(lambda t: download(*t))
ctx.execute(pipeline)

Installation

# From Marin monorepo
uv sync

# Standalone
cd lib/zephyr
uv pip install -e .

Running Tests

Zephyr tests run against multiple execution backends to ensure correctness across different environments.

All Tests on Both Backends (Default)

uv run pytest lib/zephyr/tests
# Runs all tests on both Local and Iris backends
# Local Iris cluster is started automatically via ClusterManager

Run Specific Backend Only

uv run pytest lib/zephyr/tests -k "local"
uv run pytest lib/zephyr/tests -k "iris"

The Iris cluster is started once per test session and reused across all tests for efficiency.

Design

Zephyr consolidates ad-hoc distributed and Hugging Face dataset processing patterns in Marin into a simple abstraction.

Key Features:

Lazy evaluation with operation fusion
Disk-based inter-stage data flow for low memory footprint
Chunk-by-chunk streaming to minimize memory pressure
Distributed execution with bounded parallelism (Iris/local backends)
Automatic chunking to prevent large object overhead
fsspec integration (GCS, S3, local)
Type-safe operation chaining

See AGENTS.md for execution internals and source layout.

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

marin_zephyr-0.2.96.dev33307221957.tar.gz (101.1 kB view details)

Uploaded Aug 30, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

marin_zephyr-0.2.96.dev33307221957-py3-none-any.whl (113.4 kB view details)

Uploaded Aug 30, 2026 Python 3

File details

Details for the file marin_zephyr-0.2.96.dev33307221957.tar.gz.

File metadata

Download URL: marin_zephyr-0.2.96.dev33307221957.tar.gz
Upload date: Aug 30, 2026
Size: 101.1 kB
Tags: Source
Uploaded using Trusted Publishing? Yes
Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for marin_zephyr-0.2.96.dev33307221957.tar.gz
Algorithm	Hash digest
SHA256	`c01a30c941a475473d6ac3d0483f6ad4b61cd0b02ca5ef9ae8deb6fcfc28a5a1`
MD5	`6f1735aacdadfb9ba8e5ff56a2d61168`
BLAKE2b-256	`36cac824874a3ee0cf0e04048aa5f7d1d498dcbb856bb59122d98ff91b6e708c`

See more details on using hashes here.

Provenance

The following attestation bundles were made for marin_zephyr-0.2.96.dev33307221957.tar.gz:

Publisher: marin-release-libs-wheels.yaml on marin-community/marin

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Statement:
- Statement type: https://in-toto.io/Statement/v1
- Predicate type: https://docs.pypi.org/attestations/publish/v1
- Subject name: marin_zephyr-0.2.96.dev33307221957.tar.gz
- Subject digest: c01a30c941a475473d6ac3d0483f6ad4b61cd0b02ca5ef9ae8deb6fcfc28a5a1
- Sigstore transparency entry: 2650319476
- Sigstore integration time: Aug 30, 2026
Source repository:
- Permalink: marin-community/marin@2e0bbf27ba836e26115704e06c4d4faafce8a733
- Branch / Tag: refs/heads/main
- Owner: https://github.com/marin-community
- Access: public
Publication detail:
- Token Issuer: https://token.actions.githubusercontent.com
- Runner Environment: github-hosted
- Publication workflow: marin-release-libs-wheels.yaml@2e0bbf27ba836e26115704e06c4d4faafce8a733
- Trigger Event: schedule