Build a scientific pipeline

Chain calculation, verification and save steps, or fan out across inputs in parallel.

Split multi-stage workflows into Run → Check → Save jobs so that if verification or database saving fails, you never have to re-run an expensive calculation.

1. Run the 3-stage pipeline example

Run compute/platform/examples/pipeline/run_pipeline.py to execute a 3-stage chain (run → check → save):

Tip: Using the local dev stack? Add --config .local-dev/compute/client.json.

The script prints the three job IDs, waits for the final save stage, and downloads its report:

text

Run 0195f3c0-1111-7000-8000-000000000001Check 0195f3c0-2222-7000-8000-000000000002Save 0195f3c0-3333-7000-8000-000000000003{"value":42,"project_id":"...","source_job_id":"...","check_job_id":"..."}Outputs: pipeline-output/0195f3c0-3333-7000-8000-000000000003

2. How the stages connect

First, look at the three worker functions in compute/platform/examples/pipeline/jobs.py:

python

"""Produce, check and save a small artifact-backed report bound to the job's project."""
from __future__ import annotations
import jsonfrom uuid import UUID
from pydantic import BaseModel, ConfigDict
from chemistry_compute import Collection, DataClient, job_context, one_result

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

class Input(Model):    file: str

class Measurement(Model):    value: int

class Checked(Model):    value: int    source_job_id: UUID

class Report(Model):    value: int    project_id: UUID    source_job_id: UUID    check_job_id: UUID

def run(request: Input) -> Measurement:    value = int(job_context().input(request.file).read_text()) * 2    job_context().output("measurement.txt").write_text(str(value))    return Measurement(value=value)

def check(request: Collection) -> Checked:    parent = one_result(request, Measurement, artifacts=("results/measurement.txt",))    if int(parent.artifacts["results/measurement.txt"].read_text()) != parent.value.value:        raise ValueError("measurement artifact differs from its typed result")    return Checked(value=parent.value.value, source_job_id=parent.source.job_id)

def save(request: Collection) -> Report:    parent = one_result(request, Checked)    project = DataClient().query("SELECT authz.current_project_id()::text")["rows"][0][0]    report = Report(        value=parent.value.value,        project_id=UUID(project),        source_job_id=parent.value.source_job_id,        check_job_id=parent.source.job_id,    )    job_context().output("report.json").write_text(json.dumps(report.model_dump(mode="json")))    return report

Next, look at how run_pipeline.py submits all three jobs up front using depends_on:

python

"""Run a complete dependency chain and fetch its project-bound report."""
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("pipeline-output"))    args = parser.parse_args()    client = Client.from_login(args.config)    source = Path(__file__).resolve().parent    run = client.publish(Program(jobs.run, source), f"example.pipeline.run@{args.version}")    check = client.publish(Program(jobs.check, source), f"example.pipeline.check@{args.version}")    save = client.publish(Program(jobs.save, source), f"example.pipeline.save@{args.version}")    measured = client.submit(        run, jobs.Input(file="number.txt"), inputs={"number.txt": source / "number.txt"}    )    checked = client.submit(check, Collection(), depends_on=[measured.id])    saved = client.submit(save, Collection(), depends_on=[checked.id])    print(f"Run {measured.id}\nCheck {checked.id}\nSave {saved.id}", flush=True)    print(saved.result(timeout=300).model_dump_json())    saved.fetch(args.output)    print(f"Outputs: {args.output / saved.id}")

if __name__ == "__main__":    main()

How data flows between stages:

job_context() reads staged input files (.input("number.txt")) and writes output files (.output("measurement.txt") → results/measurement.txt).
depends_on=[measured.id] tells the scheduler to hold check until measured succeeds. Once submitted, your laptop can disconnect and the pipeline still finishes on the cluster.
Collection() + one_result() gives the child job the parent's typed return value (parent.value) and local paths to its output files (parent.artifacts["results/measurement.txt"]).

For structured scientific outputs, use Parquet and reuse the producer's accepted artifact. See Share and stream scientific datasets for a complete file-producing pipeline, bounded readers and the distinction between full input staging and partial trajectory reads.

3. Fan out and reduce with a Flow

When you need to run the same calculation across many inputs in parallel and combine the results, use flows.prepare(...).map(...).reduce(...).then(...):

Here is the complete driver (compute/platform/examples/programs/run_flow.py):

python

"""Publish and run prepare → map → reduce → follow-up using your saved CLI login."""
from __future__ import annotations
import argparsefrom pathlib import Path
import jobs
from chemistry_compute import Client, Program, flows

def main() -> None:    parser = argparse.ArgumentParser(description=__doc__)    parser.add_argument("--config", type=Path)    parser.add_argument("--version", default="1")    args = parser.parse_args()    client = Client.from_login(args.config)    source = Path(__file__).resolve().parent    prepare = client.publish(Program(jobs.prepare, source), f"example.prepare@{args.version}")    double = client.publish(Program(jobs.double, source), f"example.double@{args.version}")    total = client.publish(Program(jobs.total, source), f"example.total@{args.version}")    increment = client.publish(Program(jobs.increment, source), f"example.increment@{args.version}")    flow = (        flows.prepare(prepare, jobs.Count(count=3), concurrency=2)        .map(double)        .reduce(total)        .then(increment)        .submit(client)    )    print(f"Flow {flow.id}", flush=True)    print(flow.result(timeout=300).model_dump_json())  # (0 + 2 + 4) + 1 = 7

if __name__ == "__main__":    main()

Each stage is type-checked against the next:

prepare takes Count(count=3) and returns Items[Number] (0, 1, 2).
.map(double) runs double on each item in parallel (0, 2, 4).
.reduce(total) receives Items[Number] and sums them (6). Use .collect(reducer) instead if your reducer also needs parent files or failed item records.
.then(increment) runs a follow-up step on the reduced result (7).

4. Check status and debug failures

Inspect a Flow or any individual stage from the CLI:

If a stage fails, upstream stages keep their completed outputs. Our built-in transition state and DFT workflows use this exact pattern so ct-cli ts retry and ct-cli dft retry can resume from the failed step.