Rebuilding Cloud ETL on Arrow: Polars, Lazy Frames, and Out-of-Core Jobs

A nightly job read 40 GB of parquet from object storage, joined it against three reference tables, aggregated, and wrote results back. It ran on a small Spark cluster because that is what you do with 40 GB. It now runs on one 32 GB machine in under four minutes, and the cluster is gone.

That is not a story about a faster library. It is about three things arriving together: columnar in-memory representation with no per-row Python overhead, a query optimiser that rewrites your pipeline before executing it, and a streaming engine that processes data larger than memory in chunks.

Lazy evaluation is where the wins come from

Written eagerly, a pipeline does exactly what you wrote, in order — read every column of every file, then filter, then join, then drop most of it. Written lazily, you describe the result and the optimiser reorganises the work: predicates push down into the parquet scan so unmatched row groups are never read, projections prune columns at the source, and the join order is chosen from statistics rather than from the order you typed.

import polars as pl

# Nothing executes until collect(). Everything below is a plan.
q = (
    pl.scan_parquet('s3://lake/events/date=2026-06-*/**.parquet')
      .filter(pl.col('event_type') == 'conversion')     # pushed into the scan
      .join(pl.scan_parquet('s3://lake/dim/accounts.parquet'), on='account_id')
      .group_by(['account_id', pl.col('ts').dt.truncate('1d')])
      .agg([
          pl.col('amount').sum().alias('revenue'),
          pl.col('session_id').n_unique().alias('sessions'),
      ])
      .filter(pl.col('revenue') > 0)
)

print(q.explain())              # read the plan before blaming the engine
df = q.collect(engine='streaming')   # out-of-core: memory stays bounded

Reading the plan is a habit worth building. It is where you discover that a filter did not push down because it wrapped a column in a Python function, which quietly converted a scan of two row groups into a scan of four hundred files.

One machine goes further than people assume

Distributed processing was the answer to a memory constraint that has substantially moved. A commodity cloud instance with 128 GB of RAM and fast NVMe handles datasets that genuinely required a cluster a decade ago, and it does so without shuffle overhead, scheduler latency, or an hour of your week spent tuning executor memory. For a large share of pipelines in the hundreds-of-gigabytes range, a single node is both faster end-to-end and an order of magnitude cheaper.

• Scan, do not read. scan_parquet enables pushdown; read_parquet loads everything and discards it later.
• Keep data in Arrow across process boundaries — zero-copy handoff between Python, DuckDB and your query engine removes an entire class of serialisation cost.
• Watch for Python UDFs. One map_elements in the middle of a pipeline collapses the vectorised execution you came for.
• Use the streaming engine for anything approaching memory, and test with a full-size input rather than a sample.
• Partition your lake sensibly. The optimiser can only skip files whose layout tells it something.

Most teams running a cluster are paying distributed-systems overhead to process data that fits on one machine they could rent by the hour.

The honest boundary: past a few terabytes per job, with genuinely distributed shuffles, or where you need an existing ecosystem of cluster-bound tooling, a distributed engine is still correct. Below that, the default has changed. Start with one node and an optimiser, measure, and add distribution when a measurement rather than an assumption tells you to.