Write assets in Python. rivers resolves the graph, plans execution, and runs the scheduler as compiled Rust — so nothing in the control plane waits on a Python interpreter.
pip install rivers
import rivers as rs
@rs.Asset
def raw_data():
return {"users": 100, "events": 5000}
@rs.Asset
def summary(raw_data: dict):
return f"{raw_data['users']} users"
repo = rs.CodeRepository(assets=[raw_data, summary])
repo.materialize()rivers reads the type-annotated parameters and builds the dependency graph. No wiring step.
Most orchestrators run their scheduler, planner, and daemon as Python processes. rivers compiles all of them into one binary. Python only runs inside your own asset code.
| Component | Python orchestrators | rivers |
|---|---|---|
| Graph resolution | Python | Rust (petgraph) |
| Execution planner | Python | Rust |
| Partition mapping | Python | Rust |
| Daemon: schedules, sensors, automation | Python | Rust, no GIL held |
| Web UI server | Python plus a JavaScript bundle | Rust: axum, Leptos, WASM |
| Kubernetes integration | Python launcher | Rust operator with CRDs |
| Python interpreter on the control plane | Yes, everywhere | None |
Plan times stay below a millisecond on graphs with thousands of nodes. See the full comparison
Wins and losses alike, on one machine, with the same Python and the same storage class. Each framework runs in its best configuration, and the highlighted figure wins the row.
| Measurement | Dagster | Prefect | rivers |
|---|---|---|---|
| Kubernetes control plane | |||
| install to ready for work | 25 s | 43 s | 45 s |
| memory at idle | 366 MB | 389 MB | 159 MB |
| images on disk | 399 MB | 374 MB | 244 MB |
| Graph engine | |||
| build a 100,000-asset graph | 35.06 s | 6.29 s | 718 ms |
| build a 50,000-asset graph with dependencies | 32.81 s | 3.89 s | 370 ms |
| resolve a selection over 10,000 assets | 960 ms | not supported | 8 ms |
| Run execution | |||
| one no-op asset, end to end | 95 ms | 1.09 s | 3 ms |
| no-op runs per second | 10/s | 1/s | 365/s |
| per step in one run of 5,000 | 25.5 ms | timed out | 0.5 ms |
| parallel efficiency, 512 sleeping assets | 4% | 23% | 76% |
| cancel a running run | 582 ms | not supported | 1 ms |
| Partitions | |||
| define and load 1,000,000 | 532 ms | not supported | 80 ms |
| materialize, count and list over 1,000,000 keys | 415 ms | not supported | 93 ms |
| Automation daemon | |||
| sensors held at a 1-second interval | 50 | not supported | 2,000+ |
| conditions held at a 1-second interval | 10,000 | not supported | 20,000+ |
| evaluate 500 schedules once | 2.2 ms | not supported | 0.38 ms |
| Backfill | |||
| partitions per second, 1,000 partitions | 1.5/s | not supported | 104.7/s |
| Concurrency | |||
| claims per second, 10,000 against four slots | 101/s | timed out | 753/s |
| Read path | |||
| answer a UI page over 10,000 runs | 21.1 ms | 48.3 ms | 2.0 ms |
| Logs | |||
| write and read 100,000 lines | 1.15 s | 75.53 s | 291 ms |
Measured on Apple Silicon. Full results and method
An asset is what your pipeline produces. A task is a step inside one. You wire neither — rivers reads the function parameters, builds the edges, and lineage comes for free.
import rivers as rs @rs.Assetdef sessions(events, users): return join(events, users)sessions goes hot and the downstream assets light up.On your laptop, one process runs the scheduler, your code, the database and the UI. There is no daemon to babysit, no workspace.yaml, and no docker-compose file.
Nothing changes when you go to production. The Helm chart runs the same components, split into their own pods.
Getting started$ rivers dev pipelines.analyticsNot a feature list. These are the differences you feel in week three.
Most orchestrators run one global daemon for every team. One slow sensor stalls everyone's triggers. rivers runs a daemon inside each code location, so a bad tick in one team's repo cannot touch another's.
How isolation worksWrite async def on an asset and rivers awaits it on the tokio event loop. No wrapper and no separate decorator. The parallel executor caps async fan-out separately from its subprocess workers.
Put time-window, static, and dynamic dimensions in the same partition definition, with no cap on dimension count. Cron accepts a seconds field, so partitions and schedules can tick below a minute.
PartitionsAnnotate the context with your model — AssetExecutionContext[MyConfig] — and the type is read from the annotation. No config DSL, no run-config dictionary, and real autocomplete on context.config.
Every concept is a single class with named constructors. Type rs.Executor. and the IDE lists every executor; rs.PartitionsDefinition. lists every partition kind. One import, nothing to memorise.
Give one dimension a run each and batch another inside it. per_dimension(multi_run=["date"], single_run=["region"]) is one run per date, with every region in it. No custom backfill code.
A cheap parse step and a 30 Gi join should not share one pod size. compute=rs.Compute(memory="32Gi") sizes that asset alone. Axes you leave unset inherit the executor default.
When the kernel kills a step, rivers reads the pod status, tells a real OOM from an eviction, and relaunches with more memory. 8 Gi, then 16, then 32, up to your ceiling.
RetriesSet OTEL_EXPORTER_OTLP_ENDPOINT and spans flow to your collector. Python logging and context.log bridge into the same subscriber, so per-step log capture needs no setup.
Each of these is one annotation in rivers.
@rs.Asset
def running_total(
self: rs.SelfDependency[int],
events: list,
) -> int:
prev = self.get_inner()
return (prev or 0) + len(events)An incremental asset loads its previous output through a typed parameter. There is no graph cycle, and no workaround.
@rs.Task
def double(x: int) -> int:
return x * 2
@rs.Asset.from_graph()
def doubled():
m = numbers().map(double)
return total(m.collect_stream())A graph asset is a sub-DAG of tasks, and each task is its own step. Fan out with .map(); a streaming collect starts downstream work on the first result instead of waiting for the slowest branch.
@rs.Asset
async def prices():
return await client.fetch("/prices")
@rs.Asset
def report(prices, inventory):
return {**prices, **inventory}Async assets run as event-loop tasks while sync assets run in subprocess workers, inside the same run.
Actions are the other verbs on an asset: optimize, vacuum, delete, or your own. Each one is a normal run, with logs, retries, and a run page. You can schedule it, backfill it, or start it from the UI.
Each verb declares what it does to the asset's state. An optimize run never triggers downstream, and a delete marks the asset as missing. rs.DeltaAsset has optimize, vacuum, and delete built in.
class Orders(rs.DeltaAsset):
@classmethod
def materialize(cls) -> pl.DataFrame:
return load_orders()
@rs.action(outcome=rs.Outcome.Unchanged)
@classmethod
def analyze(cls, ctx):
compute_stats(ctx.asset_name)
repo.run_action("optimize")Every page updates itself. Run state, materializations, and automation ticks arrive as they happen, so you get a live indicator instead of a refresh button.
Overview — Run health, success rate, and every asset group in one screen.
Registering a code location is one kubectl apply. The operator resolves the tag to an immutable digest, reconciles a Deployment and Service, and registers the endpoint for the UI to discover.
There is no workspace file to maintain, no "add code location" flow in the UI, and no extra process polling a hosted service for work. Delete the resource to remove it.
Install on KubernetesapiVersion: rivers.io/v1alpha1
kind: CodeLocation
metadata:
name: analytics
spec:
image: ghcr.io/acme/pipelines
tag: v0.2.0
module: pipelines.analyticsInstall the package, decorate a function, and materialize it.