Announcing Polars Cloud 0.10.0

Polars Cloud 0.10.0 is out. Here are some of the highlights:

• Stream distributed query results into Python with sink_batches().
• A new experimental distributed query planner, Miso.
• Distributed pl.collect_all().
• Experimental HDFS support for scans and Iceberg metadata on On-Prem.
• Hive-partitioned scans can skip a shuffle.

Stream Results Out of a Query with sink_batches()

Distributed queries have so far ended in one of two ways: a sink that writes to storage (sink_parquet, sink_csv, sink_ipc, sink_iceberg), or an execute() that writes the whole result to temporary storage for you to read back. However, you couldn’t apply custom logic to the results that were too large to hold in memory.

The new sink_batches() takes a Python function and calls it with each batch of results as it becomes ready:

lf.remote(ctx).sink_batches(lambda df: my_handler(df))

Each call receives a DataFrame. chunk_size controls how many rows are buffered before each call, and maintain_order=True calls the function serially instead of in parallel across workers. Note that it defaults to False here, unlike sink_batches in open source Polars, which maintains order by default. You can return True from the callback to stop the query early.

You should write the callback so that running it twice on the same batch is harmless (idempotency). Batches can be handed to it more than once, from different workers, for instance when a task is retried.

This path is considerably slower than the native sinks. It exists to give you flexibility alongside the native sinks, not to replace them.

New Experimental Opt-In Query Planner: Miso

0.10.0 ships a second distributed query planner, Miso, alongside the existing one we refer to as naive. It can be selected per query:

lf.remote(ctx).distributed(planner="miso").execute()

planner accepts "auto", "naive" and "miso", and "auto" still resolves to naive.

This is the first experimental release of Miso and currently it’s not the default, but in the near future this will become the default planner. We would like people to run it on real workloads so we can gather feedback. You can inspect the stage graph for both planners in the Dashboard. If you try it, we would love to hear how the resulting stage graphs and runtimes compare, for example through our Discord or the issue tracker in case of bugs.

Distributed pl.collect_all()

Several LazyFrames can now be submitted as a single distributed query instead of one query each. Each frame has to end in its own sink, attached with lazy=True so the sink becomes part of the plan instead of running there and then:

events = pl.scan_parquet("s3://bucket/events/")

queries = [
    events.filter(pl.col("kind") == "click").sink_parquet("s3://bucket/clicks/", lazy=True),
    events.group_by("user").len().sink_parquet("s3://bucket/per_user/", lazy=True),
]

pl.collect_all(queries, lazy=True).remote(ctx).distributed().execute()

Note that a frame without a sink is rejected, since there is no single result to hand back.

Common subplan elimination runs over the combined plan, so the scan shared by both frames above is read once instead of once per frame.

HDFS Support for On-Prem

On-Prem clusters can now use HDFS as a storage back-end. Workers reach hdfs://host:port/... paths through a pure-Rust client, so the cluster needs no JVM, and scan_iceberg can read Iceberg metadata over HDFS through a pyiceberg fsspec integration. Once a cluster has it enabled, data access is plain Polars:

pl.scan_parquet("hdfs://localhost:9005/data/foods1.parquet").remote(ctx).execute()

Access always happens from the workers, never from the client. Both HDFS and the Iceberg metadata path are off by default and take some cluster setup: a worker.extras toggle, the Hadoop configuration directory available on the workers, hdfs-native and pyiceberg in the worker environment for metadata access, and a few specific storage_options for Iceberg tables. The HDFS configuration guide covers all of it, Kerberos included. We are interested in hearing from anyone running Polars On-Prem against an HDFS-backed lake.

Hive-Partition Aware Scans

The distributed planner runs operations like join and group_by partitioned by key. When a scan reads Hive-partitioned data, 0.10.0 recognizes the existing partitioning and can run those operations without introducing a shuffle, which speeds up queries that previously paid for one.

Get Started

The full list of changes is in the 0.10.0 release notes on GitHub.

Try it out on Polars Cloud, or check the On-Prem releases to upgrade a self-hosted deployment to 0.7.1.

Follow us on your favorite platform for updates:

[LinkedIn] - [Twitter/X] - [Bluesky] - [Reddit (NEW)] - [GitHub]