# 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:

1. Write and summarize a Parquet dataset in a two-stage pipeline.
2. Share output files between jobs without re-uploading them.
3. 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:

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

Use a saved [CLI connection](/start/first-job/) and a CPU worker running the shared environment. From the repository root, run the pipeline:

```bash
compute/runtime/.venv/bin/python compute/platform/examples/files/run.py --output ./file-output
```

If you are using the [local development stack](/operate/local-development/), add `--config .local-dev/compute/client.json`.

The script prints the job IDs and the calculated summary:

```text
{"rows":4,"total":6.0}
```

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:

```bash
ct-cli status PRODUCER_JOB_UUID --json
ct-cli fetch PRODUCER_JOB_UUID --output ./file-output
```

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](/operate/recovery/) for handling interrupted runs.

## 3. Write bounded batches and share the accepted file

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 pa
from 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 argparse
from 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](/start/pipelines/)).

## 4. Stage existing inputs and collect outputs

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

```bash
ct-cli submit analysis.py --input measurements.parquet=./measurements.parquet --wait
```

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

```python
path = job_context().input("measurements.parquet")
with parquet.open_file(path) as document:
    ...
```

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:

```bash
ct-cli fetch JOB_UUID --output ./job-output
```

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:

```bash
compute/runtime/.venv/bin/python compute/platform/examples/files/read.py frames TRAJECTORY_UUID
```


```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:

```bash
compute/runtime/.venv/bin/python compute/platform/examples/files/read.py download FILE_UUID ./measurements.parquet
```


```python
from pathlib import Path

from chemistry_compute import parquet
from 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.