Skip to content

Remote and distributed execution

Keyten can serialize a lazy plan, execute it in a headless worker process, and stream the result back in the same block format used by native tables. This surface is available now but remains under active development.

What can ship

A remote plan must be built from file scans. A LazyFrame created from an in-memory DataFrame owns process-local data and cannot serialize.

import keyten as kt

kt.DataFrame([
    kt.Series.string("sym", ["a", "b", "a"]),
    kt.Series.int("qty", [1, 2, 3]),
]).write_native("dist_demo.k10dir")

query = (
    kt.scan_native("dist_demo.k10dir")
    .group_by("sym")
    .agg(kt.col("qty").sum().alias("total"))
    .sort("sym")
)
wire = query.serialize()
assert kt.LazyFrame.deserialize(wire).collect().to_dict()["total"] == [4, 2]

serialize() is a versioned transport format, not a long-term storage format. Worker and coordinator engine versions must match.

Start a worker

The Python wheel does not install the worker executable. Build it from the same source revision as the wheel or coordinator:

cargo build --release --bin keyten-worker
./target/release/keyten-worker --addr 127.0.0.1:7101

Useful worker options:

--addr HOST:PORT       address to listen on
--workers N            local engine threads for this worker
--map FROM=TO          rewrite a coordinator scan-path prefix on this worker

Every worker must be able to read the scanned files. Use the same mounted path everywhere, or pass --map /coordinator/path=/worker/path when mount points differ.

Use a trusted network

The worker protocol is raw TCP with no authentication or TLS. Bind it to localhost or a private, access-controlled network. Do not expose a worker port directly to the internet.

Execute remotely

collect_on sends the whole serializable plan to one worker:

# not-runnable: requires a running worker with access to dist_demo.k10dir
result = query.collect_on("127.0.0.1:7101")

collect_distributed accepts several worker addresses:

# not-runnable: requires two running workers with access to dist_demo.k10dir
workers = ["10.0.0.11:7101", "10.0.0.12:7101"]
result = query.collect_distributed(workers)

With one address, the whole plan runs on that worker. With several addresses, Keyten looks for an aggregation it can split over a native scan. sum, count, min, max, mean, and var can merge from partial results. Aggregates that need every value together, including std, median, quantile, n_unique, first, last, and corr, do not decompose.

If a serializable plan cannot split, Keyten runs the whole plan on the first worker. If the plan itself cannot serialize, collect_distributed falls back to local execution. The answer remains exact in every route.

Native scans are the current row-partitioned source. CSV and Parquet plans can ship to one worker, but are not divided across the fleet yet.

Progress and cancellation

Pass a QueryCtx to aggregate progress from the fleet and forward cancellation to every worker:

# not-runnable: requires running workers
ctx = kt.QueryCtx()
result = query.collect_distributed(workers, ctx)
stage, stages, blocks_done, blocks_total, rows_out = ctx.progress()
assert ctx.is_cancelled() is False

In a notebook, keyten.jupyter.watch(query, workers=workers) displays the optimized plan and a live fleet-wide progress bar. Interrupting the notebook calls ctx.cancel() and workers stop at their next block boundary.

Operational checklist

  • Run the same engine build on coordinator and workers.
  • Keep scanned native tables visible at matching or mapped paths.
  • Sort explicitly when result order matters.
  • Keep worker ports private.
  • Treat the plan wire format as short-lived transport, not persisted data.