Distributed writing

Each worker receives a producer-sized record batch, writes one immutable shard, and returns its small descriptor:

The root below is a local directory so that the example runs anywhere. Nothing here is specific to local storage: pass a gs:// or s3:// root instead and the same code publishes to object storage.

import numpy as np
import signal_dataset as sds

ROOT = "fleet.sds"


def batch(prefix: str, count: int) -> list[sds.Record]:
    """Stand-in for whatever the producer hands each worker."""
    return [
        sds.Record(
            id=f"{prefix}-{index:04d}",
            fields={
                "samples": sds.Field(
                    np.zeros(1024, dtype=np.complex64),
                    axes=(sds.Axis("time", 1024),),
                )
            },
            metadata={"sample_rate_hz": 1_000_000},
        )
        for index in range(count)
    ]


def worker(work_id: str, records: list[sds.Record]) -> sds.PublishedShard:
    return sds.write_shard(records, ROOT, work_id=work_id, attempt=0)

The coordinator publishes only after every expected worker reports. expected_work_ids is what makes that a check rather than a hope: publication fails if a worker never reported, instead of quietly publishing a short dataset.

work_ids = ("worker-000", "worker-001")
shards = [
    worker(work_ids[0], batch("north", 3)),
    worker(work_ids[1], batch("south", 2)),
]

dataset = sds.publish(
    ROOT,
    shards,
    dataset_id="captures",
    snapshot_id="run-001",
    expected_work_ids=work_ids,
)

assert len(dataset) == 5

Retries use the same work_id, a higher attempt, and new immutable objects. Publication selects one winner per work ID, validates the pinned object size of each shard, writes paged manifests and the snapshot, then creates root.json last. Record counts are not verified unless you ask for it with PublicationOptions(verify_record_counts=True), because that check re-reads every shard being published.

Batch size is producer-controlled. Optional StorageOptions guardrails can enforce a deployment budget; no shard-size or dataset-size ceiling is enabled by default.