Jobs & Executors¶

Jobs¶

A Job defines a named subset of assets and tasks to execute together. Jobs must be added to a CodeRepository, which resolves the dependency graph and validates the job's contents before execution.

import rivers as rs

@rs.Asset
def asset_a() -> int:
    return 1

@rs.Asset
def asset_b(asset_a: int) -> int:
    return asset_a + 1

repo = rs.CodeRepository(
    assets=[asset_a, asset_b],
    jobs=[
        rs.Job(name="my_pipeline", assets=[asset_a, asset_b], executor=rs.Executor.in_process()),
    ],
)
repo.get_job("my_pipeline").execute()

repo.get_job() returns the repository's own resolved copy of the job. The Job you construct stays a declaration, so one Job can be shared by several repositories, and each runs it with its own resources, IO handlers, executor, and retry policies.

Job.execute() returns a RunResult — check result.success, result.run_id, result.materialized_assets, result.failed_assets. Read materialized values back via repo.load_node("asset_b") (which goes through the asset's IO handler).

Subgraph validation¶

When a CodeRepository is constructed, every job is validated:

All assets must exist in the repository.
No broken chains: every upstream dependency of a job's assets must also be in the job (or marked as incomplete — see below).
Independent assets are fine: assets with no dependency relationship can coexist in the same job.

# Valid — a and c are independent
rs.Job(name="ok", assets=[a, c])

# Invalid — b depends on a, but a is not in the job
rs.Job(name="broken", assets=[b, c])

Incomplete dependencies¶

Set allow_incomplete_deps=True to allow missing upstream dependencies. A missing dependency that an asset loads as an input needs an io_handler that can load its data. A lineage-only dependency (AssetDef.dep) loads nothing, so it needs none:

rs.Job(name="partial", assets=[b], allow_incomplete_deps=True)

Use this when upstream assets are materialized separately. repo.materialize(selection) applies the same rule to the assets outside the selection.

Partitioned execution¶

Pass a PartitionKey to materialize a specific partition:

repo.get_job("my_pipeline").execute(
    partition_key=rs.PartitionKey.single("2024-01-15"),
)

Executors¶

An Executor picks how a job's steps are dispatched — for materialize runs; an action job's steps always run in the run's own process. Ctrl-C stops the steps that run in the process: a sync step gets KeyboardInterrupt, an async step asyncio.CancelledError. Construct one through the static factories on Executor:

InProcess¶

Runs sync steps one at a time in the calling Python process, and up to four async steps at the same time. Best for debugging and small pipelines:

executor = rs.Executor.in_process()

Parallel (subprocess pool)¶

Runs sync steps concurrently in subprocesses (loky pool) and async steps concurrently as tasks:

executor = rs.Executor.parallel(max_workers=4)

Argument	Default	Notes
`max_workers`	`os.cpu_count()`	Subprocess pool size for sync steps.
`max_async_concurrent`	unbounded	Cap on concurrent async tasks.

Note

Executor.parallel() does not cloudpickle your asset bodies. Functions and IO handlers are sent to workers as (module, qualname) refs and re-imported in the subprocess — they need to be importable on the worker's Python path, not pickle-friendly. Upstream inputs are loaded inside the worker via the IO handler (so input values never cross the boundary), outputs are written from the worker, and each Resource is rebuilt in the worker from its field values, secrets included, with its own setup() and teardown(). Closures and locally-defined functions (<locals>) fall back to direct pickling — keep production assets at module scope.

Kubernetes¶

Each step runs as its own Kubernetes worker pod:

executor = rs.Executor.kubernetes(
    worker_image="my-registry/rivers:1.2.3",
    max_concurrent_steps=20,
    namespace="rivers",
    service_account="rivers-executor",
    worker_cpu="500m",
    worker_memory="512Mi",
)

worker_image defaults to the controlling pod's image when omitted. See the Executors API reference for every parameter.

Default executor¶

When no default_executor is specified on CodeRepository, the default is Executor.parallel(). Jobs without an explicit executor inherit this default.

Per-asset executor override¶

Individual assets can override the executor via the rivers/executor metadata key. This works in both Job.execute() and CodeRepository.materialize():

@rs.Asset(metadata={"rivers/executor": "in_process"})
def needs_in_process() -> int:
    return 42

Valid values: "in_process", "parallel". See Executors API for details.