Scale a query

A remote query runs on the distributed engine by default: one query spread over several workers, so it can process data larger than the memory of one machine.

Set up

The deployment sets the size of the worker pool, the number of workers a query uses when it does not request a count of its own, and the per-query maximum.

AWSHelm chartBare-metal

cluster_size on the compute context sets the number of machines, which is both the default and the per-query maximum. cpus and memory, or instance_type, set the size of each machine, as described in Compute context definition.

ctx = pn.ComputeContext(
    workspace="My workspace",
    cpus=8,
    memory=32,
    cluster_size=4,
)

worker.deployment.replicaCount sets the number of worker pods (default 2) and, on a fixed-size cluster, both the default and the per-query maximum. workersPerQuery.default and workersPerQuery.max override those two.

worker:
  deployment:
    replicaCount: 4
workersPerQuery:
  default: 2

For worker memory, see Worker resources and memory. To grow and shrink the pool with demand, see Autoscaling.

The worker pool is the worker nodes you start: every node with the worker component enabled joins the pool when it registers with the scheduler. scheduler.default_workers_per_query sets the default, and scheduler.max_workers_per_query the per-query maximum. Without a default, a query that sets no maximum of its own can use up to max_workers_per_query, or every worker when that is also unset.

[scheduler]
enabled = true
default_workers_per_query = 4
max_workers_per_query = 8

For worker memory, see Worker resources and memory. The Bare-metal config file reference lists the keys. To grow and shrink the pool with demand, see Autoscaling.

Single node or distributed

.distributed() on the result of lf.remote(ctx) runs the query on the distributed engine. It is the default, so the call is only needed to pass arguments such as min_workers or max_workers. .single_node() runs the whole query on one worker instead:

lf.remote(ctx).distributed().collect()

lf.remote(ctx).single_node().collect()

The distributed engine suits a query whose working set (after predicate and projection pushdown) does not fit in the RAM of the largest single machine available.

A query that fits is not always fastest on one machine. In a benchmark at a similar price, scan-bound queries ran faster on many small machines, which add read bandwidth. Join-heavy queries ran faster on one large machine, where .single_node() avoids shuffles. A shuffle redistributes data across workers between stages and writes intermediate results to disk.

When the data fits comfortably on the local machine, the open-source Polars library can run the query locally with lf.collect(), and the LazyFrame needs no changes.

Even in a distributed query, some stages run on one worker: a map_batches call without is_elementwise=True, a scan of data created in the client such as pl.LazyFrame({...}), and memory-intensive operations on small inputs. The query profiler marks these stages.

Workers per query

min_workers and max_workers on distributed() set the number of workers a query uses. Both are optional and can be given on their own. The example asks for at least 2 and at most 8 workers:

lf.remote(ctx).distributed(min_workers=2, max_workers=8).collect()

Without max_workers, the ceiling is the cluster's default number of workers per query from Set up. The planner splits the query into partitions for that ceiling, even when fewer workers are present at submit time.

Setting	Behavior
Neither	Starts as soon as one worker is available, and grows to the ceiling.
`min_workers` only	Waits for that many workers, then grows to the ceiling.
`max_workers` only	Starts as soon as one worker is available, and grows to at most that many.
Both	Waits for `min_workers`, then uses up to `max_workers`.

A request above the cluster's per-query maximum is capped without an error. A query whose min_workers cannot be met fails once a fixed-size pool has not changed for 3 minutes or an autoscaled pool reaches its scale-up timeout.

A worker runs one query at a time, and a query claims idle workers up to its maximum. So queries submitted together run side by side only when their max_workers leave workers free. Otherwise, a later query waits for workers.

Measure memory and runtime

There is no fixed ratio between input size and memory: a filter streams through the data while a group_by, join or sort holds state. Run the query at a small scale factor of the public datasets first, and read the memory-intensive indicator and the shuffle volume in the query profile.

When a worker runs out of memory, first try lf.remote(ctx).distributed(partitions_per_worker=4): smaller tasks lower the peak memory, and idle workers pick up the remaining tasks. If that is not enough, or one memory-intensive stage dominates the runtime, give the workers more memory or add workers (see Set up).

Autoscaling

An autoscaled cluster scales its worker pool while it runs, instead of holding a fixed number of workers. Autoscaling is available on the Helm chart and on bare-metal, but AWS compute uses a fixed pool of the size set in Set up. The scheduler scales the pool to the worker demand of the running queries, between a minimum and a maximum size.

Demand is the sum of the worker counts requested by all running queries, so two concurrent queries that each want four workers produce a demand of eight. A query that does not request a count of its own uses the cluster's default number of workers per query. An autoscaled cluster must set this default, because it is the capacity each such query requests and the number of workers the query is planned for.

The scheduler asks for more workers as soon as demand rises, and asks for fewer once the pool has been idle for 60 seconds. It never scales below the number of workers that currently hold a query.

Helm chartBare-metal

scaling.enabled: true autoscales the pool between scaling.minReplicas (default 0) and scaling.maxReplicas (unbounded when unset). With scaling enabled, the chart refuses to render without workersPerQuery.default. The pool only scales down while no query is running, because a Deployment gives the scheduler no way to choose which pod is removed. The scale-up timeout is fixed at 3 minutes.

scaling:
  enabled: true
  maxReplicas: 8
workersPerQuery:
  default: 2

Autoscaling needs exactly one of [scaling.rest] or [scaling.grpc] on the scheduler node, for a scaling service that you run. The scaling service provisions and removes the machines and reports the pool's minimum and maximum size.

In that section, scale_up_timeout sets the scale-up timeout (5 minutes by default): the window the scaling service has to provision workers. For a scaling service that cannot choose which workers to remove, scale_down_on_idle_only = true makes the pool scale down only while no query is running. An autoscaled scheduler refuses to start without default_workers_per_query.

[scheduler]
enabled = true
default_workers_per_query = 4

[scaling.rest]
uri = "http://<SCALING_SERVICE_HOST>:8080"