Job document#

A job document is a JSON object describing one job. Only moruna_spec, source and sink are required; every other field has the same default as the matching moruna.run argument. Fields that are absent or null are filled from the environment, then from the machine, then from the default. Unknown fields are refused.

{
  "moruna_spec": 1,
  "source": {"kind": "parquet", "url": "reviews/"},
  "kernels": [{"kind": "python", "module": "kernels.py", "callable": "score"}],
  "sink": {"kind": "parquet", "url": "reviews-scored/"},
  "budget": {"memory_bytes": 8589934592}
}

Top-level fields#

Field	Type	Default	Meaning
`moruna_spec`	integer	required	The document format version; `1`.
`run_id`	string	generated	The job’s id, 32 lowercase hexadecimal characters.
`source`	object	required	Where the input comes from.
`kernels`	list	`[]`	The kernels, in order; empty copies the source to the sink.
`sink`	object	required	Where the output goes.
`budget`	object	discovered	Memory and CPU.
`staging`	object	discovered	The directory for spilled data and checkpoints.
`object_store`	object	none	Connection settings for object storage URLs.
`checkpoint`	object	enabled, every 5 s	Checkpoint settings.
`resume`	string	none	`"auto"`, a run id or the path to a `manifest.json`.
`error_policy`	string or object	`"terminate"`	`"terminate"`, `"skip"` or `{"budget": n}`.
`ordered`	boolean	`false`	Deliver results to the sink in source order.
`sizer`	string	`"rule"`	`"rule"` or `"learned"`.
`trace`	string	none	Path of a file for the per-morsel trace.
`profiles_dir`	string	`~/.moruna/profiles`	Where kernel profiles are stored.
`allow_gil`	boolean	`false`	Run Python kernels on the standard build of Python.
`report`	object	report file only	Where the report goes.
`contracts`	list	`[]`	The contracts a `datafusion` source reads and a `peql` sink writes under, compiled.

source#

kind selects the source; the other fields depend on it.

parquet: url is a path, a URL or a list of them. options.columns lists the columns to read, and options.filters lists row-group filters as [column, op, value], with op >, < or ==.

{"kind": "parquet", "url": "s3://acme-data/reviews/", "options": {"columns": ["review_id", "text"]}}

vortex: url is a path, a URL or a list of them; a directory or prefix is read for its .vortex files. options.columns lists the columns to read, and options.split_bytes sets the size each piece of work is cut toward (default 128 MiB).

{"kind": "vortex", "url": "s3://acme-data/events/", "options": {"columns": ["event_id", "text"]}}

tensor: url is a path or a list of paths to safetensors or aligned binary tensor files; options.tensors lists the tensors to read.

datafusion: a query planned by peQL for a caller. root is the peQL workspace, where peQL keeps its manifests, privacy ledger and audit log; the contracts are the ones listed in contracts; exactly one of contract, to read one contract, or sql, a query over contracts; and caller, the caller as peQL reads it, with id, tenant and purpose and optionally tier, clearance, classification, roles, now and other. on_violation says what the read does with rows that break a contract’s row rules: exclude (default) leaves them out, and the run report’s read counts them per rule for each contract; refuse refuses the read when the data holds any, naming each rule and how many rows broke it. See the peQL documentation.

{"kind": "datafusion", "root": "/data/workspace", "contract": "purchasing/orders",
 "caller": {"id": "svc-scoring", "tenant": "acme", "purpose": "reporting"}}

An iterator source exists only in Python and is refused in a document.

kernels#

Each entry has a kind.

python: module is a module name or a path to a .py file, and callable is the name of the kernel in it, dotted for a nested name. The decorator’s options can be given here as well: stateful, instances, accepts, tier, device_memory, releases_gil, expected_amplification, preferred_rows, resume, state_bytes and memory_guard.

std: name is a standard kernel, such as "filter", and args is an object of its arguments.

{"kind": "std", "name": "select", "args": {"columns": ["review_id", "score"]}}

Any entry can set fingerprint, as printed by moruna check. Moruna then refuses to run the job if the loaded kernel’s fingerprint differs, so a document cannot run a kernel that has changed since it was checked.

sink#

parquet: url is a directory or prefix. options.row_group_bytes (default 128 MiB), options.file_bytes (default 1 GiB) and options.compression: zstd (default), snappy, gzip, lz4 or none.

vortex: url is a directory or prefix. options.file_bytes (default 1 GiB) is the most a file may hold.

tensor: url is a directory. options.format is mrb1 (default) or safetensors; options.one_file_per_morsel and options.name (default tensor).

arrow_ipc: url is a directory; options.file_bytes (default 1 GiB).

peql: writes under a peQL contract listed in contracts. root is the peQL workspace, contract the contract, caller the writer, who must own the contract, and mode either append (default) or overwrite. peQL writes the data in the contract’s layout and updates the contract’s manifest when the job finishes.

on_violation says what the write does with rows that break the contract’s row rules:

refuse (default): nothing is committed when any row breaks a rule, and the job fails naming each rule and how many rows broke it.
exclude: the rows that pass are committed, and the others are written as Parquet under excluded_url, one file per morsel that had any, each row with a _violations column listing the rules it broke. excluded_url is a local directory, given only with exclude, that is empty or absent when the job starts and is apart from the contract’s files and from the source.
warn: every row is committed, and peQL’s verdict on the write marks it invalid.

The run report’s write says what was done: report.violations (rows by rule), report.rows_excluded and, for exclude, excluded with the directory, the files and their rows.

{"kind": "peql", "root": "/data/workspace", "contract": "purchasing/orders",
 "caller": {"id": "svc-ingest", "tenant": "acme", "purpose": "ingest"},
 "on_violation": "exclude", "excluded_url": "/staging/excluded"}

contracts#

Each entry is one contract, as peQL registers a compiled one: document, the contract’s document; compiled, its compiled form exactly as parcel wrote it; functions, the function modules the compiled form pins; and audiences, the tenants it is published to for this job besides its owner (public is everyone). The job’s peQL engine holds these contracts in memory and nothing else: it reads no contract from root and writes none there, and a contract the job names that is not listed here is refused by name. A document that lists contracts and has no datafusion source or peql sink is refused.

budget#

Field	Meaning
`memory_bytes`	The memory limit for the whole process, in bytes.
`cpu`	The number of cores.
`elastic.memory_max_bytes`	The largest memory limit the job may grow to if the machine gains memory.
`elastic.cpu_max`	The largest number of cores the job may grow to.

Without elastic, the job keeps the limits it started with. See Machines that change size.

staging#

Field	Meaning
`dir`	An existing directory for spilled data and checkpoints.
`limit_bytes`	The most disk the job may use there; default 20% of the free space.
`durable`	`true` if the directory is on a disk that outlives the machine.

object_store#

s3 takes endpoint, region, access_key_id, secret_access_key, session_token and bucket. gcs takes service_account_path, service_account_json and bucket. azure takes account, access_key and container. allow_http permits endpoints without TLS, and local_root sets the directory file:// URLs are relative to. An endpoint can be a unix:// or vsock:// socket, for a host that provides object storage without a network.

checkpoint#

Field	Default	Meaning
`enabled`	`true`	Write checkpoints while the job runs.
`interval_ms`	`5000`	Milliseconds between checkpoints, from 500 to 60,000.
`keep`	`false`	Keep the checkpoint directory after the job completes.

report#

Field	Meaning
`file`	Where to write the report; `<run_id>` in the path is replaced with the job’s id. By default, `moruna-<run_id>.report.json` in the staging directory, or in the current directory.
`socket`	A `unix://` or `vsock://` address to connect to and send progress and the report; see the host protocol.