Share and stream scientific datasets

Write Parquet datasets, share accepted files between jobs and choose between partial reads and full downloads.

Scientific workflows produce datasets too large to store in database rows or pass around as JSON. In chemistry.tools, you write structured data to Parquet artifacts, pass lightweight references between jobs, and let workers or web clients stream or download only what they need.

In this guide, you will:

Write and summarize a Parquet dataset in a two-stage pipeline.
Share output files between jobs without re-uploading them.
Choose between streaming partial windows and downloading whole files.

1. Use Parquet for structured data

Choose Parquet by default for measurement tables, trajectories, and per-item benchmark results:

Data type	Recommended representation	Why
Trajectories, coordinates, measurements	Parquet with explicit column types and units	Preserves schemas; internal Zstd page compression enables fast range queries
Job parameters and return values	Typed Pydantic models (or small JSON)	Lightweight, human-readable, and validated at entry
Raw scripts, checkpoints, images	Original native format	Preserves the exact binary required by external tools

Use chemistry_compute.parquet to read and write files. It handles encryption and internal Zstd compression automatically.

Note: Never wrap .parquet files in .gz or .zst. Parquet compresses individual column chunks internally. Compressing the outside of the file destroys seekability and prevents byte-range streaming.

Keep full scientific fidelity in your columns. Parquet's column-level compression makes datasets compact without needing to truncate precision, drop coordinates, or reduce sampling rates.

2. Run a file-producing pipeline

The example in compute/platform/examples/files/ demonstrates file handling using two jobs:

Producer: Generates four synthetic measurements and saves them to results/measurements.parquet.
Consumer: Reads that accepted artifact from the producer and writes results/summary.json.

Use a saved CLI connection and a CPU worker running the shared environment. From the repository root, run the pipeline:

If you are using the local development stack, add --config .local-dev/compute/client.json.

The script prints the job IDs and the calculated summary:

The consumer downloads summary.json under file-output/SUMMARY_JOB_UUID/results/. The original dataset remains attached to the producer job; the consumer does not upload a duplicate copy. To inspect or download that original file:

If a job fails, inspect its output with ct-cli logs JOB_UUID. If only the consumer stage fails, retain the producer's ID and submit a new consumer that depends on it. See run recovery for handling interrupted runs.

Here are the producer and consumer functions from compute/platform/examples/files/jobs.py:

python

"""Write a synthetic Parquet dataset and summarize its accepted artifact in another job."""
from __future__ import annotations
import pyarrow as pafrom pydantic import BaseModel, ConfigDict, Field
from chemistry_compute import Collection, job_context, one_result, parquet

class Model(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True, allow_inf_nan=False)

class Samples(Model):    count: int = Field(default=4, ge=0)

class Dataset(Model):    rows: int

class Summary(Model):    rows: int    total: float

def produce(request: Samples) -> Dataset:    schema = pa.schema(        [("sample_id", pa.int64()), ("value", pa.float64())],        metadata={b"value_unit": b"dimensionless", b"source": b"synthetic example"},    )    path = job_context().output("measurements.parquet")    with parquet.writer(path, schema) as output:        for start in range(0, request.count, 1024):            indices = list(range(start, min(start + 1024, request.count)))            table = pa.table(                {"sample_id": indices, "value": [float(i) for i in indices]}, schema=schema            )            output.write_table(table, row_group_size=1024)    return Dataset(rows=request.count)

def summarize(request: Collection) -> Summary:    parent = one_result(request, Dataset, artifacts=("results/measurements.parquet",))    rows, total = 0, 0.0    with parquet.open_file(parent.artifacts["results/measurements.parquet"]) as document:        for batch in document.iter_batches(columns=["value"], batch_size=1024):            rows += batch.num_rows            total += sum(batch.column(0).to_pylist())    if rows != parent.value.rows:        raise ValueError("dataset row count differs from its typed result")    summary = Summary(rows=rows, total=total)    job_context().output("summary.json").write_text(summary.model_dump_json(indent=2) + "\n")    return summary

Key patterns in the code:

Writing in batches: job_context().output("measurements.parquet") creates a path under results/. For large datasets, use parquet.writer(...) and write bounded batches rather than accumulating every row in memory.
Finishing the file: Exiting the with block closes the writer and flushes the Parquet footer. Always close writers before your job returns.
Reading upstream artifacts: one_result(request, Dataset, artifacts=(...)) validates that the parent job succeeded and locates its staged Parquet file on the worker.

Connecting stages in the submission script:

python

"""Submit a Parquet producer and consumer using the saved CLI connection."""
from __future__ import annotations
import argparsefrom pathlib import Path
import jobs
from chemistry_compute import Client, Collection, Program

def main() -> None:    parser = argparse.ArgumentParser(description=__doc__)    parser.add_argument("--config", type=Path)    parser.add_argument("--version", default="1")    parser.add_argument("--output", type=Path, default=Path("file-output"))    args = parser.parse_args()    client = Client.from_login(args.config)    source = Path(__file__).resolve().parent    produce = client.publish(Program(jobs.produce, source), f"example.files.produce@{args.version}")    summarize = client.publish(        Program(jobs.summarize, source), f"example.files.summarize@{args.version}"    )    produced = client.submit(produce, jobs.Samples(count=4))    summarized = client.submit(summarize, Collection(), depends_on=[produced.id])    print(f"Produce {produced.id}\nSummarize {summarized.id}", flush=True)    print(summarized.result(timeout=300).model_dump_json())    summarized.fetch(args.output)    print(f"Outputs: {args.output / summarized.id}")

if __name__ == "__main__":    main()

Setting depends_on=[produced.id] ensures the consumer only runs after the producer finishes with accepted outputs.

Because the jobs are linked, the consuming worker stages the parent artifact directly from storage into its working directory. Your local machine never has to download and re-upload the intermediate file just to pass it onward.

For pipelines with multiple producers, use results(request, Dataset) to iterate over a Collection, or use a Flow's .collect(...) reducer (see Build a scientific pipeline).

4. Stage existing inputs and collect outputs

To pass a local file from your machine into a remote job, declare it as an input:

Inside analysis.py, access the local staged path with job_context().input("measurements.parquet"):

Only explicitly declared inputs are uploaded; surrounding directory files are not included automatically.

Collecting outputs

Any file your script writes inside results/ is automatically uploaded and verified when the job finishes. Use ct-cli fetch or Python's job.fetch() to download them:

Downloads are verified against stored digests and replace destination files atomically.

5. Choose partial reads or a full download

The platform provides two distinct ways to access files:

Task	Recommended path	How it works
Preview a few frames, scrub a trajectory, plot metrics	Stream a window	Requests only the byte ranges for the selected steps
Full analysis, model training, bulk parsing	Download the file	Streams the entire file to disk once, then processes locally
Read specific columns from a local dataset	Column projection	Opens local file and decodes only requested columns

Iterating in batches keeps memory usage low while processing data locally—it does not mean making hundreds of separate network calls. For bulk analysis across an entire run, download the file once and stream through it locally.

Inspect a trajectory window

For published, artifact-backed MD trajectories, inspect specific simulation steps without downloading the full dataset:

python

from chemistry_compute.remote import Connection, RemoteClient

def read_window(api: RemoteClient, project_id: str | None, trajectory_id: str) -> None:    page = api.frames(        project_id,        trajectory_id,        {"selection": {"kind": "window", "start": 0, "stride": 1, "limit": 20}, "forces": False},    )    for frame in page["rows"]:        print(frame["step"], frame["potential_energy_ev"])    print("Next window starts at:", page["next_step"])

The reader fetches only the necessary column chunks and footer ranges directly from storage. Use next_step from the response to page through subsequent windows (None indicates the end of the trajectory).

In browser workspaces, the 3D viewer uses this exact streaming mechanism inside a Web Worker to scrub through trajectories smoothly without downloading gigabytes of coordinates. Note that streaming reads completed, accepted artifacts—it cannot tail an active simulation while it is still running.

Download once for whole-file processing

For bulk analysis or training where you need to examine every row or frame, download the complete artifact:

python

from pathlib import Path
from chemistry_compute import parquetfrom chemistry_compute.remote import Connection, RemoteClient

def download_file(api: RemoteClient, file_id: str, destination: Path) -> None:    api.fetch(file_id, destination)    rows = 0    with parquet.open_file(destination, resolve_key=api.read_key) as document:        for batch in document.iter_batches(batch_size=1024):            rows += batch.num_rows    print(f"Read {rows} rows from {destination}")

api.fetch(...) streams the file directly to disk and verifies its integrity. If the artifact was encrypted with native Parquet encryption, supply resolve_key=api.read_key when opening it. Inside a worker job, helper utilities resolve keys automatically.

6. Let the platform handle storage

When you use the shared chemistry_compute clients, the platform manages storage transparently:

Direct transfers: Artifact bytes travel directly between workers or clients and S3/R2 storage via presigned URLs. Large file transfers never bottleneck through the central backend server.
End-to-end encryption: Artifacts are encrypted before storage. Authorized clients resolve access keys on demand through the API; keys are never stored in plain URLs.
Transparent chunking: Large files (over ~240 MiB) are split into bounded storage objects for reliable caching and parallel transfer, but they behave as a single logical file with continuous offsets.