Polars Streaming Engine Deep Dive: Out-of-Core Joins, GroupBy, Window

polars streaming is the mode that turns Polars from a fast-DataFrame-in-RAM library into a legitimate TB-scale local ETL engine — capable of processing datasets that don't fit in memory by streaming batches of rows through a chain of Rust-fused expressions and spilling to disk when a hash partition or sort buffer exceeds available RAM. Every Python data engineer eventually reaches for Polars for local ETL; knowing when to use collect(streaming=True) versus .collect() versus .sink_parquet(), when over() still works in streaming mode versus when the engine falls back, and when to reach for DuckDB SQL instead of Polars DataFrames is what separates a mid-level operator from a senior one.

The tour walks the five pillars — (1) the lazy vs eager vs streaming mode decision (eager for interactive < 10 GB, lazy for query planning, streaming for out-of-core), (2) out-of-core joins via external merge and group-bys via partitioned hash spill, plus sink_parquet for zero-materialisation output, (3) window functions in streaming mode with over() and rolling_mean(window_size=N) for time-series, (4) expression fusion — the Polars trick that collapses chained expressions into single Rust passes for streaming perf, and (5) the decision matrix comparing Polars streaming to DuckDB (SQL, better join planner) and Pandas (interactive, do NOT use > 10 GB). Every section ships a Solution-Tail interview answer — code, trace, output, why-this-works with __concept__ underlines.

Practice on SQL library →, SQL optimization drills →, and SQL aggregation drills →.

---

On this page

• Why Polars streaming matters in 2026
• Lazy vs eager vs streaming mode
• Out-of-core joins and group-bys
• Window functions in streaming
• Patterns + comparison + dialect matrix
• Cheat sheet — Polars streaming recipe list
• Frequently asked questions
• Practice on PipeCode

---

1. Why Polars streaming matters in 2026

The polars streaming mental model — process TB-scale data on a laptop by streaming batches through fused Rust expressions and spilling only when necessary

The one-sentence invariant: Polars' streaming engine executes a LazyFrame plan in bounded-memory batches — reading input in chunks, applying fused expressions on each chunk, spilling hash partitions or sort buffers to disk when they exceed RAM, and either producing an in-memory DataFrame (.collect(streaming=True)) or writing directly to Parquet without ever materialising the full result (.sink_parquet()) — enabling a 500 GB parquet-to-parquet ETL on a laptop with 16 GB RAM where Pandas would OOM and Spark would need a cluster.

Where Polars streaming actually shows up.

• Local TB-scale ETL — process a 500 GB parquet directory on a laptop before committing to a warehouse spend.
• Batch jobs that must not blow past RAM — an Airflow task with a 4 GB container limit needs to process 100 GB of CSV.
• CSV → parquet conversion at scale — flatten large CSV archives into columnar format for downstream compute.
• CI data validation on full production dumps — nightly job scans full events table via streaming, validates counts, publishes report.
• Feature engineering — join events with dim tables, compute window features, sink to parquet.
• Data quality auditing — scan warehouse dumps for null rate spikes, cardinality drops, referential integrity breaks.

When to reach for streaming vs eager.

• Streaming when data > 50% of available RAM OR when the pipeline is batch (not interactive).
• Eager when data < 25% RAM AND you want the fastest possible execution (no streaming overhead) OR you're doing interactive analysis (Jupyter).
• Lazy always for building the query — the terminal call (.collect() vs .collect(streaming=True) vs .sink_parquet()) chooses execution mode.

What senior interviewers probe when Polars comes up.

• When to use streaming vs eager. Data-size vs RAM ratio.
• sink_parquet vs collect(streaming=True). Zero-materialisation output vs streaming to in-memory result.
• Out-of-core join. External merge or hash spill; how it works and its cost.
• Expression fusion. Chained ops fuse into single Rust pass; critical for streaming perf.
• When streaming fails. Complex window functions, deeply nested expressions.
• Polars vs DuckDB. Both are Rust; DuckDB is SQL-first with a stronger join optimiser.
• Polars vs Pandas. Polars 5-30× faster and handles > 10 GB; Pandas still best for interactive < 10 GB.

The four Polars mental models a senior needs.

• LazyFrame is a plan, not data. pl.scan_csv("big.csv") reads nothing; .filter(...) adds a filter node to the plan; .collect() executes.
• Expression fusion is compile-time. The plan is compiled to a single Rust program before execution; chained expressions become a single pass.
• Streaming is chunk-based. The engine reads chunks (default 1 M rows), applies plan per chunk, streams outputs.
• Sink is one-way to disk. sink_parquet streams straight to Parquet without a materialised in-memory result — the killer feature for TB-scale.

Worked example — Pandas OOM vs Polars streaming success

Detailed explanation. A 200 GB CSV needs to be filtered, joined with a dim table, and written to Parquet. Pandas: pd.read_csv tries to materialise; OOM. Polars streaming: reads in chunks, spills join hash when needed, sinks to Parquet.

Question. Show both approaches; explain why Polars wins.

Input. events.csv (200 GB, 2B rows) + dim_users.parquet (5 M rows). Task: filter events to last 90 days, join with dim, write to enriched.parquet.

Code.

# PANDAS — will OOM
import pandas as pd
events = pd.read_csv("events.csv")            # OOM here
dim = pd.read_parquet("dim_users.parquet")
recent = events[events.event_date >= "2026-04-01"]
enriched = recent.merge(dim, on="user_id")
enriched.to_parquet("enriched.parquet")

# POLARS STREAMING — succeeds
import polars as pl
(
    pl.scan_csv("events.csv")
      .filter(pl.col("event_date") >= pl.date(2026, 4, 1))
      .join(pl.scan_parquet("dim_users.parquet"), on="user_id")
      .sink_parquet("enriched.parquet")
)

Step-by-step explanation.

1. Pandas read_csv allocates 200 GB in RAM — laptop with 16 GB dies immediately.
2. Polars scan_csv returns a LazyFrame — reads nothing, builds a plan node.
3. filter adds a filter node to the plan.
4. join adds a join node; the join is streaming-eligible since the right side (5 M dim) fits in memory as a hash table.
5. sink_parquet triggers execution — reads events in 1 M-row chunks, filters, joins each chunk against the in-memory dim hash, writes chunks to Parquet.
6. Peak RAM ~1-2 GB (chunk buffer + dim hash + output buffer).
7. Wall clock ~8-15 min on a laptop.

Output.

Approach	Peak RAM	Wall clock	Success
Pandas	OOM at read	—	fail
Polars streaming	~2 GB	~10 min	success

Rule of thumb. Any dataset > 25% of RAM should default to Polars lazy or streaming — even if it would fit in eager.

Worked example — the streaming vs eager decision

Question. For a 5 GB CSV → aggregate by day, which mode?

Code.

import polars as pl

# EAGER — good; fits in RAM
df = pl.read_csv("5gb.csv")
result = df.group_by("date").agg(pl.col("revenue").sum())

# LAZY EAGER — same, but with plan optimisation
result = (
    pl.scan_csv("5gb.csv")
      .group_by("date")
      .agg(pl.col("revenue").sum())
      .collect()   # default: not streaming
)

# STREAMING — unnecessary for 5 GB on 32 GB laptop; slight overhead
result = pl.scan_csv("5gb.csv").group_by("date").agg(pl.col("revenue").sum()).collect(streaming=True)

Rule of thumb. Eager for data < 25% RAM; streaming for > 50%; lazy for query planning; sink_parquet for zero-materialisation.

Worked example — expression fusion in action

Detailed explanation. Polars compiles chained expressions into a single Rust pass; without fusion, streaming a 6-op chain would need 6 passes over the data.

Code.

# 4 chained expressions
lf = (
    pl.scan_parquet("data.parquet")
      .with_columns([
          (pl.col("price") * pl.col("qty")).alias("total"),
          pl.col("ts").dt.month().alias("month"),
      ])
      .filter(pl.col("total") > 100)
      .group_by("month")
      .agg(pl.col("total").sum())
)

# Polars compiles into a single Rust pass:
#   - read chunk
#   - project price*qty, ts.month
#   - filter total > 100
#   - hash aggregate by month
#   - emit chunk

print(lf.explain())   # prints the physical plan

Rule of thumb. Chain expressions in one pipeline call; Polars fuses them; multiple .collect() calls in a row break fusion.

Common beginner mistakes

• Using eager read_csv on large files — allocates all of RAM.
• Assuming Pandas syntax works — Polars uses .filter(pl.col("x") > 0) not df[df.x > 0].
• Calling .collect() in the middle of a chain — breaks streaming fusion.
• Not using scan_* — the streaming benefit starts with lazy reads.
• Using streaming on data that fits in RAM — extra overhead for no gain.
• Wide SELECT * on Parquet — no column projection pushdown.

polars streaming interview question on choosing between Polars, DuckDB, and Pandas

A senior interviewer often opens with: "You have a 500 GB parquet directory, need to filter, join with 3 dim tables, aggregate by date, write result. You have a 32 GB laptop. Choose the tool and justify."

Solution Using Polars streaming for the fan-in + sink_parquet for zero-materialisation

import polars as pl

result_path = "enriched_daily.parquet"

pipeline = (
    pl.scan_parquet("events/*.parquet")
      .filter(pl.col("event_date") >= pl.date(2026, 4, 1))
      .join(pl.scan_parquet("dim_users.parquet"), on="user_id")
      .join(pl.scan_parquet("dim_products.parquet"), on="product_id")
      .join(pl.scan_parquet("dim_regions.parquet"), on="region_id")
      .with_columns([
          pl.col("event_date").dt.truncate("1d").alias("day"),
      ])
      .group_by(["day", "region_name", "product_category"])
      .agg([
          pl.col("revenue").sum().alias("revenue"),
          pl.col("user_id").n_unique().alias("dau"),
      ])
      .sink_parquet(result_path)
)

Step-by-step trace.

Step	Action	Memory
1	scan_parquet — plan node, no I/O	~0
2	filter — plan node, pushdown to reader	~0
3	3 joins — dim tables loaded once as in-memory hashes	~500 MB
4	with_columns — expression added	~0
5	group_by + agg — streaming hash aggregate	~1 GB
6	sink_parquet — writes chunks; no full result in RAM	~1 GB total

Peak RAM ~2 GB (chunk buffers + dim hashes + partial aggregate).

Output:

Choice	Fit	Reason
Polars streaming + sink_parquet	✅ best fit	Handles 500 GB on 32 GB laptop
DuckDB	good alternative	SQL-first; similar perf; better join optimiser
Pandas	✗ fails	OOM on read
Spark	overkill for laptop	Adds cluster complexity

Why this works — concept by concept:

• sink_parquet for zero-materialisation — the killer feature that makes 500 GB fit on 32 GB.
• Filter pushdown — Polars pushes the date filter into the Parquet reader; only reads matching row groups.
• Dim tables in RAM as hashes — small dims fit; large fact streams through.
• Expression fusion — the entire pipeline compiles to one Rust program; single pass per chunk.
• Hash aggregate spill — if grouping cardinality exceeds RAM, Polars partitions the aggregate to disk.
• Cost — wall clock ~15-30 min on modest laptop; peak RAM ~2 GB. On a cloud VM, similar; the point is you don't need a cluster.

SQL
Topic — aggregation
SQL aggregation drills

Practice →

SQL Topic — optimization SQL optimization drills

Practice →

---

2. Lazy vs eager vs streaming mode

pl.scan_* builds a plan; .collect() materialises; .collect(streaming=True) streams; .sink_* never materialises

The mental model in one line: every Polars operation happens in one of three modes — eager (data materialised in RAM immediately per operation, best for interactive analysis under 25% RAM), lazy (build a query plan without executing, materialise once with .collect(), best for query optimisation), and streaming (execute the plan in chunks with disk spill, best for data > 50% RAM); sink_* is a special terminal that streams straight to disk without ever holding the full result.

Slot 1 — eager mode.

import polars as pl

df = pl.read_csv("data.csv")            # materialises immediately
df2 = df.filter(pl.col("x") > 0)       # new DataFrame in RAM
df3 = df2.group_by("k").agg(pl.col("v").sum())

• Each op returns a materialised DataFrame.
• Best for interactive analysis.
• Simplest debugging.
• Data must fit in RAM.

Slot 2 — lazy mode.

result = (
    pl.scan_csv("data.csv")
      .filter(pl.col("x") > 0)
      .group_by("k")
      .agg(pl.col("v").sum())
      .collect()
)

• scan_csv returns LazyFrame (no data read).
• Each op adds to the plan.
• .collect() executes the whole plan optimally.
• Enables predicate pushdown, projection pushdown, expression fusion.

Slot 3 — streaming mode.

result = (
    pl.scan_csv("data.csv")
      .filter(pl.col("x") > 0)
      .group_by("k")
      .agg(pl.col("v").sum())
      .collect(streaming=True)
)

• Same plan as lazy.
• collect(streaming=True) executes chunk-by-chunk.
• Spills to disk when needed.
• Result still materialised (in memory) — for large results, use sink_*.

Slot 4 — sink terminals.

(
    pl.scan_parquet("events/*.parquet")
      .filter(...)
      .group_by(...)
      .agg(...)
      .sink_parquet("output.parquet")   # streams direct to disk
)

• .sink_parquet(path) — Parquet output.
• .sink_csv(path) — CSV.
• .sink_ipc(path) — Arrow IPC.
• .sink_ndjson(path) — newline-delimited JSON.
• Never holds full result in RAM.

Slot 5 — scan_* readers.

• pl.scan_csv(path, ...) — CSV.
• pl.scan_parquet(path, ...) — Parquet (best for streaming).
• pl.scan_ipc(path, ...) — Arrow IPC.
• pl.scan_ndjson(path, ...) — JSON lines.
• Globs supported: pl.scan_parquet("events/*.parquet").

Slot 6 — plan inspection.

lf = pl.scan_csv("big.csv").filter(pl.col("x") > 0).group_by("k").agg(pl.col("v").sum())
print(lf.explain())            # logical plan
print(lf.explain(optimized=True))  # optimised plan
print(lf.explain(streaming=True))  # streaming physical plan

Slot 7 — predicate pushdown.

result = (
    pl.scan_parquet("events/*.parquet")
      .filter(pl.col("date") == "2026-07-12")
      .collect()
)
# Polars pushes the filter into the Parquet reader; only reads matching row groups.

Confirmed by .explain() showing the filter applied at read.

Slot 8 — projection pushdown.

result = (
    pl.scan_parquet("events/*.parquet")
      .select(["user_id", "revenue"])
      .collect()
)
# Only reads user_id and revenue columns from Parquet.

Slot 9 — when streaming beats eager, even for in-RAM data.

• Streaming may still be slightly slower per unit data.
• But keeps memory bounded — safer for shared CI environments.
• Use eager for one-off scripts; streaming for production ETL.

Slot 10 — Polars streaming isn't Spark streaming.

• Polars "streaming" = out-of-core batch processing.
• Not real-time / event-driven.
• For streaming Kafka events, use aiokafka + Polars for periodic batch aggregation.

Common beginner mistakes

• Using read_csv on large files — allocates full RAM.
• Calling .collect() in the middle of a pipeline — breaks fusion.
• Using streaming when data fits in RAM — unnecessary overhead.
• Forgetting sink_* for output — the streaming benefit dies if you materialise then write.
• Not using .explain() to verify pushdown.

Worked example — the mode decision

Question. Given a 500 MB CSV, a 50 GB CSV, and a 500 GB CSV directory, pick the mode for each.

Data size	RAM	Mode
500 MB	32 GB	Eager (read_csv) — fits easily
50 GB	32 GB	Lazy + collect (~all in RAM after ops) OR streaming for safety
500 GB	32 GB	Streaming with sink_parquet (must — won't fit)

Rule of thumb. Data > 25% RAM → streaming; else eager or lazy.

Worked example — explain output

Code.

lf = (
    pl.scan_parquet("events.parquet")
      .filter(pl.col("date") >= pl.date(2026, 6, 1))
      .select(["user_id", "revenue"])
      .group_by("user_id")
      .agg(pl.col("revenue").sum())
)
print(lf.explain(streaming=True))

Output (simplified):

STREAMING:
  Aggregate [user_id]: [sum(revenue)]
    Project [user_id, revenue]
    Filter [date >= 2026-06-01]
    Parquet scan: events.parquet
      predicate pushdown: date >= 2026-06-01
      projection pushdown: [user_id, revenue, date]

Rule of thumb. Always .explain() before running large streaming jobs to verify pushdown.

Worked example — collect vs sink

Code.

# collect(streaming=True) — result in RAM
result: pl.DataFrame = lf.collect(streaming=True)

# sink_parquet — result on disk, never fully in RAM
lf.sink_parquet("result.parquet")
result: pl.DataFrame = pl.read_parquet("result.parquet")   # if needed

Rule of thumb. For TB-scale output, sink; for small aggregate result, collect(streaming=True).

polars streaming interview question on choosing the mode for a pipeline

A senior interviewer asks: "Design a pipeline that reads 200 GB of parquet daily, joins with a 500 MB dim, aggregates, writes 10 MB result. Which mode and which terminal?"

Solution Using scan + streaming + sink_parquet for the ETL; collect(streaming=True) for the aggregate

# ETL: stream 200 GB → aggregate → 10 MB result

lf = (
    pl.scan_parquet("events/*.parquet")
      .join(pl.scan_parquet("dim.parquet"), on="user_id")
      .group_by(["day", "region"])
      .agg([pl.col("revenue").sum(), pl.col("user_id").n_unique().alias("dau")])
)

# Since output is small (10 MB), collect(streaming=True) is fine
result = lf.collect(streaming=True)
result.write_parquet("daily_agg.parquet")   # or use sink_parquet in pipeline

Why this works — concept by concept:

• scan_parquet for fact table — never loads all 200 GB.
• Small dim table joined via hash — 500 MB fits in RAM.
• Streaming group_by — hash aggregate spills if group cardinality is large.
• collect(streaming=True) for small result — 10 MB aggregate fits; materialise as DataFrame.
• Peak RAM ~2 GB — chunk buffer + dim hash + partial aggregate.

SQL
Topic — aggregation
SQL aggregation drills

Practice →

SQL Topic — SQL SQL practice library

Practice →

---

3. Out-of-core joins and group-bys

sink_parquet for zero-materialisation output, external merge join, and partitioned hash spill for group-by

The mental model in one line: when the Polars streaming engine encounters an operation whose intermediate result exceeds available RAM — a join build side, a hash aggregate table, a sort buffer — it partitions the operation across disk-backed batches; the join uses either an in-memory hash (small side fits) or external merge (both sides large), the group-by uses partitioned hash spill, and sink_parquet streams the final result direct to Parquet without any full-result materialisation.

Slot 1 — sink_parquet.

(
    pl.scan_parquet("events/*.parquet")
      .filter(pl.col("date") >= pl.date(2026, 6, 1))
      .group_by("user_id")
      .agg(pl.col("revenue").sum())
      .sink_parquet("result.parquet")
)

• Never holds full result in RAM.
• Writes chunks as they complete.
• Peak RAM = chunk buffer + partial aggregate.

Slot 2 — in-memory join (small right side).

(
    pl.scan_parquet("events.parquet")   # 500 GB fact
      .join(pl.scan_parquet("dim.parquet"), on="id")   # 100 MB dim
      .sink_parquet("out.parquet")
)

• Polars loads dim as hash table (100 MB).
• Streams fact through, probing dim hash per row.
• Peak RAM = dim hash + chunk buffer.

Slot 3 — external merge join (both sides large).

• When both sides too big for hash in RAM.
• Sort both by join key (external merge sort).
• Zip merged streams.
• Slower than hash join but survives RAM limits.
• Polars picks automatically based on statistics.

Slot 4 — partitioned hash group-by.

• Group-by cardinality × avg group size may exceed RAM.
• Polars partitions by hash of group key into N buckets.
• Each bucket processed independently; spilled if needed.
• Merged for final result.

Slot 5 — spill directory.

• Default: system temp dir.
• Override: POLARS_TMPDIR=/large-disk polars at process start.
• Ensure spill dir has enough space (2-3× the data being processed).

Slot 6 — sink variants.

• sink_parquet(path, compression="snappy"|"zstd"|"lz4").
• sink_csv(path, separator=",").
• sink_ipc(path) — Arrow IPC.
• sink_ndjson(path) — JSON lines.

Slot 7 — join strategy hints.

Polars picks join algorithm automatically; can nudge via .join(..., how="inner"|"left"|"outer"|"semi"|"anti", on=..., strategy="hash"|"sort") (strategy hint added in recent versions).

Slot 8 — join key must match dtype.

• int32 vs int64 on join key → Polars raises.
• Cast explicitly: pl.col("user_id").cast(pl.Int64).

Slot 9 — anti-join for "not in" pattern.

active = pl.scan_parquet("active.parquet")
churned = pl.scan_parquet("churned.parquet")
still_active = active.join(churned, on="user_id", how="anti").sink_parquet("out.parquet")

Slot 10 — cross join warning.

• pl.scan_a().join(pl.scan_b(), how="cross") — materialises N×M rows.
• Almost never what you want at scale.

Common beginner mistakes

• Joining without checking dtypes — silent failure or crash.
• Using cross join by accident (missing on= in some engines' inference).
• Not sinking — using collect for large output → OOM.
• Not verifying spill dir has enough space.
• Assuming external merge is fast — it's much slower than in-memory hash.

Worked example — daily aggregate on 500 GB fact table

Code.

(
    pl.scan_parquet("events/*.parquet")
      .filter(pl.col("date") >= pl.date(2026, 6, 1))
      .group_by(["date", "region"])
      .agg([
          pl.col("revenue").sum(),
          pl.col("user_id").n_unique().alias("dau"),
          pl.col("session_id").count().alias("sessions"),
      ])
      .sort(["date", "region"])
      .sink_parquet("daily_region.parquet")
)

Rule of thumb. Group-by cardinality (dates × regions ~1000) is small; hash aggregate fits in RAM easily; streaming reads chunks and updates the aggregate.

Worked example — join 5 dim tables to a 200 GB fact

Code.

(
    pl.scan_parquet("fact.parquet")
      .join(pl.scan_parquet("dim_user.parquet"), on="user_id")
      .join(pl.scan_parquet("dim_product.parquet"), on="product_id")
      .join(pl.scan_parquet("dim_region.parquet"), on="region_id")
      .join(pl.scan_parquet("dim_channel.parquet"), on="channel_id")
      .join(pl.scan_parquet("dim_date.parquet"), on="date")
      .sink_parquet("enriched.parquet")
)

Each dim loaded once as hash; fact streams through.

Rule of thumb. Star schema fits streaming perfectly — small dims in RAM, large fact streaming.

Worked example — anti-join for churn detection

Code.

active_users = (
    pl.scan_parquet("events_this_month.parquet")
      .select("user_id").unique()
)

churned = (
    pl.scan_parquet("all_users.parquet")
      .join(active_users, on="user_id", how="anti")
      .sink_parquet("churned.parquet")
)

Rule of thumb. Anti-join = LEFT JOIN + WHERE IS NULL in SQL; Polars makes it explicit.

polars streaming interview question on partitioned join

A senior interviewer asks: "You're joining two 100 GB parquet tables on a shared key. Neither fits in RAM. What does Polars do and what should you do?"

Solution Using pre-partition + join per partition + concat

# Approach 1: Rely on Polars external merge (may be slow)
result = pl.scan_parquet("a.parquet").join(
    pl.scan_parquet("b.parquet"), on="key"
).sink_parquet("out.parquet")

# Approach 2: Pre-partition by key, join per partition, concat
import polars as pl

for partition_id in range(64):
    (
        pl.scan_parquet("a.parquet")
          .filter(pl.col("key_hash") == partition_id)
          .join(
              pl.scan_parquet("b.parquet").filter(pl.col("key_hash") == partition_id),
              on="key",
          )
          .sink_parquet(f"out_part_{partition_id}.parquet")
    )

# Downstream: read out_part_*.parquet via glob
result = pl.scan_parquet("out_part_*.parquet")

Why this works — concept by concept:

• Pre-partitioning by hash of key — each partition fits in RAM.
• Per-partition join — in-memory hash join per partition; fast.
• Concat via glob — Polars stitches back.
• Cost — extra pass to partition; but avoids the slow external merge sort.

SQL
Topic — joins
SQL join drills

Practice →

SQL Topic — SQL SQL practice library

Practice →

---

4. Window functions in streaming

over() + rolling_mean(window_size=N) + cum_sum — streaming-compatible window operations

The mental model in one line: Polars window functions come in three forms — expr.over(partition_col) for SQL-style window aggregates (equivalent to SUM() OVER (PARTITION BY col)), rolling_*(window_size=N) for time-series-style moving windows, and cum_* for cumulative — most work in streaming mode as long as the partition doesn't exceed RAM, but very high-cardinality or wide-partition windows fall back to non-streaming and require chunking manually.

Slot 1 — over() — SQL-style window.

(
    pl.scan_parquet("orders.parquet")
      .with_columns([
          pl.col("total").sum().over("customer_id").alias("customer_total"),
          pl.col("total").rank(descending=True).over("customer_id").alias("rank_by_total"),
      ])
      .sink_parquet("out.parquet")
)

Equivalent SQL:

SELECT
    *,
    SUM(total) OVER (PARTITION BY customer_id) AS customer_total,
    RANK() OVER (PARTITION BY customer_id ORDER BY total DESC) AS rank_by_total
FROM orders;

Slot 2 — cumulative functions.

• pl.col("x").cum_sum() — cumulative sum.
• pl.col("x").cum_max() — cumulative max.
• pl.col("x").cum_min().
• pl.col("x").cum_count().
• Add .over("group_col") for per-group cumulative.

Slot 3 — rolling functions.

(
    pl.scan_parquet("timeseries.parquet")
      .sort("timestamp")
      .with_columns([
          pl.col("value").rolling_mean(window_size=7).alias("mean_7"),
          pl.col("value").rolling_sum(window_size=30).alias("sum_30"),
          pl.col("value").rolling_std(window_size=7).alias("std_7"),
      ])
      .sink_parquet("windowed.parquet")
)

Slot 4 — time-based rolling.

(
    pl.scan_parquet("timeseries.parquet")
      .sort("timestamp")
      .with_columns([
          pl.col("value").rolling_mean_by(
              by="timestamp",
              window_size="7d",
          ).alias("mean_7d"),
      ])
      .sink_parquet("windowed.parquet")
)

Windows by time interval, not row count.

Slot 5 — expression fusion in streaming.

Chained expressions compile into a single Rust pass per chunk:

lf = (
    pl.scan_parquet("data.parquet")
      .with_columns([
          (pl.col("price") * pl.col("qty")).alias("total"),
          pl.col("ts").dt.date().alias("day"),
      ])
      .filter(pl.col("total") > 100)
      .group_by("day")
      .agg([pl.col("total").sum()])
)

Slot 6 — when window functions break streaming.

• Very wide partitions (billions of rows per group).
• rank() requires sort — expensive at scale.
• shift() with large lag — buffer overflow.
• Some complex window compositions fall back to non-streaming.

Verify with .explain(streaming=True) — streaming plan or fallback?

Slot 7 — ranking variants.

• pl.col("x").rank() — standard.
• pl.col("x").rank(method="dense") — dense (no gaps on ties).
• pl.col("x").rank(method="ordinal") — ordinal (arbitrary tie-break).
• pl.col("x").rank(descending=True).

Slot 8 — top-N per group.

(
    pl.scan_parquet("orders.parquet")
      .with_columns(
          pl.col("total").rank(descending=True).over("customer_id").alias("rn")
      )
      .filter(pl.col("rn") <= 3)
      .sink_parquet("top3_per_customer.parquet")
)

Top 3 orders per customer.

Slot 9 — lag / lead.

pl.col("value").shift(1).over("group_col").alias("prev")
pl.col("value").shift(-1).over("group_col").alias("next")

Slot 10 — combining aggregate over with plain aggregate.

(
    pl.scan_parquet("orders.parquet")
      .with_columns([
          pl.col("total").sum().over("customer_id").alias("customer_total"),
      ])
      .filter(pl.col("customer_total") > 1000)   # customers with > $1K total
      .sink_parquet("high_value.parquet")
)

Common beginner mistakes

• .over() on ungrouped column — Polars needs the partition column.
• Rolling functions without sort — undefined result.
• Time-based rolling without datetime type on the by column.
• Assuming all window functions stream — verify with .explain().
• Using rank on very-high-cardinality data without partitioning.

Worked example — 7-day rolling revenue per customer

Code.

(
    pl.scan_parquet("events.parquet")
      .filter(pl.col("event_date") >= pl.date(2026, 6, 1))
      .sort(["customer_id", "event_date"])
      .with_columns([
          pl.col("revenue").rolling_sum_by(
              by="event_date",
              window_size="7d",
          ).over("customer_id").alias("rev_7d"),
      ])
      .sink_parquet("rolling.parquet")
)

Rule of thumb. Time-based rolling by customer requires sort per customer + time-based window — expensive but correct.

Worked example — top-1 order per customer

Code.

(
    pl.scan_parquet("orders.parquet")
      .with_columns(
          pl.col("total").rank(descending=True, method="ordinal").over("customer_id").alias("rn")
      )
      .filter(pl.col("rn") == 1)
      .drop("rn")
      .sink_parquet("top1_per_customer.parquet")
)

Rule of thumb. Top-N per group in Polars = rank over partition + filter.

Worked example — cumulative sum per session

Code.

(
    pl.scan_parquet("events.parquet")
      .sort(["session_id", "ts"])
      .with_columns(
          pl.col("value").cum_sum().over("session_id").alias("running_total")
      )
      .sink_parquet("running.parquet")
)

Rule of thumb. Cumulative per group = cum_* over partition + sort inside.

polars streaming interview question on rolling window over TB-scale time series

A senior interviewer asks: "Compute 30-day rolling average and 7-day rolling stddev on a 500 GB time-series parquet, per sensor. Explain memory and disk implications."

Solution Using sort by sensor + time + rolling_*_by

(
    pl.scan_parquet("sensors/*.parquet")
      .sort(["sensor_id", "ts"])
      .with_columns([
          pl.col("value").rolling_mean_by(
              by="ts", window_size="30d",
          ).over("sensor_id").alias("mean_30d"),
          pl.col("value").rolling_std_by(
              by="ts", window_size="7d",
          ).over("sensor_id").alias("std_7d"),
      ])
      .sink_parquet("rolling.parquet")
)

Why this works — concept by concept:

• Sort by sensor + ts — required for .over("sensor_id") to produce correct rolling windows.
• External merge sort — Polars sorts sensor-time on disk if it doesn't fit in RAM.
• Rolling per partition — each sensor's data processed independently after sort.
• Sink to Parquet — no full-result materialisation.
• Peak RAM — sort buffer + rolling window per partition ~few GB.
• Disk usage — spill dir needs 2-3× data size during sort.

SQL
Topic — window-functions
Window function drills

Practice →

SQL Topic — aggregation SQL aggregation drills

Practice →

---

5. Patterns + comparison + dialect matrix

Polars streaming vs DuckDB vs Pandas — when each wins

The mental model in one line: for local TB-scale ETL in Python, Polars streaming and DuckDB are the two dominant choices — Polars is DataFrame-first with expression fusion in Rust and streaming that spills to disk; DuckDB is SQL-first with a stronger cost-based optimiser for complex joins and better SQL semantics — they interop through Arrow and Parquet with zero copy, so many teams use both, while Pandas remains the interactive-analysis default for data under 10 GB.

Slot 1 — Polars strengths.

• Rust-native expression fusion.
• Streaming out-of-core.
• DataFrame API preferred by Python engineers.
• sink_* for zero-materialisation output.
• Compatible with Arrow, Parquet, IPC.

Slot 2 — DuckDB strengths.

• SQL-first (better for teams comfortable with SQL).
• Stronger cost-based optimiser (complex joins).
• In-process (embedded); no server.
• Native support for Parquet, CSV, JSON, Iceberg, Delta.
• Async support since 0.10.

Slot 3 — Pandas strengths.

• Ecosystem (matplotlib, sklearn, statsmodels).
• Interactive REPL.
• Best for < 10 GB.
• Immense community + tutorials.

Slot 4 — the interop story.

• Polars → Arrow → DuckDB with zero copy.
• DuckDB → Arrow → Polars with zero copy.

# Polars → DuckDB
import duckdb
result_df = duckdb.sql("SELECT * FROM polars_df WHERE x > 0").pl()

# DuckDB → Polars
duckdb.sql("SELECT * FROM 'data.parquet'").pl()

Slot 5 — the comparison table.

Task	Polars streaming	DuckDB	Pandas
500 GB parquet aggregation	8 min	12 min	OOM
Interactive analysis	Awkward (lazy)	SQL-friendly	Best
Complex SQL	Awkward	Best	N/A
Rust perf	✅	✅	❌
Streaming out-of-core	✅ (native)	✅ (partial)	❌
Ecosystem	Growing	Growing	Massive

Slot 6 — when streaming fails on Polars.

• Very complex window functions on huge partitions.
• Some deeply nested expressions.
• Fallback: process by chunk manually with collect(streaming=False) per chunk.
• Or drop to DuckDB SQL for the problem step.

Slot 7 — polyglot pipeline pattern.

# Ingest with Polars streaming
(
    pl.scan_csv("raw/*.csv")
      .filter(...)
      .sink_parquet("staging.parquet")
)

# Complex SQL join with DuckDB
duckdb.sql("""
    COPY (
        SELECT ...
        FROM staging JOIN dim ON ...
        WHERE ...
        GROUP BY ...
    ) TO 'result.parquet' (FORMAT PARQUET)
""")

# Interactive exploration with Pandas
df = pd.read_parquet("result.parquet")   # small enough now

Slot 8 — Polars migration from Pandas.

• df[df.x > 0] → df.filter(pl.col("x") > 0).
• df.groupby("k").sum() → df.group_by("k").agg(pl.all().sum()).
• df["col"] = expr → df.with_columns(expr.alias("col")).
• df.merge(other, on="k") → df.join(other, on="k").
• df.to_csv("out.csv") → df.write_csv("out.csv").

Slot 9 — Polars migration from DuckDB.

Both interop via Arrow; migration is usually optional (use both).

Slot 10 — when Pandas is still best.

• Data < 10 GB.
• Interactive Jupyter analysis.
• ML/statistics ecosystem integration.
• Team knows Pandas.

Common beginner mistakes

• Assuming Polars is a Pandas drop-in — syntax differs significantly.
• Not using scan_* — starting eager on large data.
• Ignoring DuckDB — sometimes SQL is simpler.
• Using Polars for < 1 GB where Pandas is faster (setup overhead).

Worked example — polyglot pipeline

Code.

import polars as pl
import duckdb

# 1. Ingest raw CSVs with Polars streaming
(
    pl.scan_csv("raw/*.csv")
      .filter(pl.col("date") >= pl.date(2026, 1, 1))
      .sink_parquet("staging.parquet")
)

# 2. Complex join + agg with DuckDB
duckdb.sql("""
    COPY (
      SELECT c.region, SUM(o.total) AS revenue, COUNT(DISTINCT o.customer_id) AS dau
      FROM 'staging.parquet' o
      JOIN 'dim_customer.parquet' c ON c.id = o.customer_id
      WHERE o.status = 'shipped'
      GROUP BY c.region
    ) TO 'result.parquet' (FORMAT PARQUET)
""")

# 3. Load result for downstream use with Polars
result = pl.read_parquet("result.parquet")
print(result)

Rule of thumb. Use each tool where it shines; interop via Parquet is free.

Worked example — Pandas → Polars migration

Code.

# Pandas
import pandas as pd
df = pd.read_csv("data.csv")
df["total"] = df["price"] * df["qty"]
result = df[df.total > 100].groupby("region")["total"].sum().reset_index()

# Polars equivalent
import polars as pl
result = (
    pl.scan_csv("data.csv")
      .with_columns((pl.col("price") * pl.col("qty")).alias("total"))
      .filter(pl.col("total") > 100)
      .group_by("region")
      .agg(pl.col("total").sum())
      .collect(streaming=True)
)

Rule of thumb. Polars chains fluently; Pandas uses assignment; both express same semantics.

Worked example — Polars → DuckDB → Polars

Code.

import polars as pl
import duckdb

# Polars for initial transform
pl_df = pl.scan_parquet("events.parquet").filter(pl.col("date") > pl.date(2026, 6, 1)).collect()

# DuckDB for complex SQL join
result = duckdb.sql("""
    SELECT o.customer_id, COUNT(*) AS orders
    FROM pl_df o
    GROUP BY o.customer_id
""").pl()   # returns Polars DataFrame via zero-copy Arrow

# Back to Polars for downstream
print(result.filter(pl.col("orders") > 10))

Rule of thumb. Zero-copy Arrow interop makes polyglot pipelines cheap.

polars streaming interview question on when to reach for DuckDB instead

A senior interviewer asks: "You have a 200 GB parquet with a 6-way join across 5 dim tables and a complex GROUPING SETS aggregate. Would you use Polars or DuckDB, and why?"

Solution Using DuckDB for the complex SQL, Polars for pre/post transforms

import duckdb

duckdb.sql("""
    COPY (
        SELECT
            c.region,
            p.category,
            r.rep_name,
            SUM(o.revenue) AS revenue,
            COUNT(DISTINCT o.customer_id) AS dau,
            GROUPING(c.region, p.category, r.rep_name) AS gid
        FROM 'events.parquet' o
        JOIN 'dim_customer.parquet' c ON c.id = o.customer_id
        JOIN 'dim_product.parquet' p ON p.id = o.product_id
        JOIN 'dim_region.parquet' r ON r.id = o.region_id
        JOIN 'dim_channel.parquet' ch ON ch.id = o.channel_id
        JOIN 'dim_date.parquet' d ON d.date = o.date
        WHERE o.date >= '2026-01-01'
        GROUP BY GROUPING SETS (
            (c.region, p.category, r.rep_name),
            (c.region, p.category),
            (c.region),
            ()
        )
    ) TO 'result.parquet' (FORMAT PARQUET)
""")

Why this works — concept by concept:

• DuckDB for complex SQL — GROUPING SETS + 6-way join is more expressive in SQL than Polars DataFrame API.
• Cost-based optimiser — DuckDB has a stronger optimiser for multi-way joins.
• Streaming out-of-core — DuckDB streams too; comparable memory profile to Polars.
• Parquet in/out — no copy; direct file scan.
• Interop — result Parquet readable by Polars, Pandas, downstream tools.

SQL
Topic — SQL
SQL practice library

Practice →

SQL
Topic — joins
SQL join drills

Practice →

---

Cheat sheet — Polars streaming recipe list

• pl.scan_* — LazyFrame from file (parquet, csv, ipc, ndjson).
• .collect() — materialise in RAM.
• .collect(streaming=True) — streaming engine, result in RAM.
• .sink_parquet(path) — stream direct to Parquet (no materialisation).
• .sink_csv / .sink_ipc / .sink_ndjson — same for other formats.
• Expression fusion — chained ops compile to single Rust pass.
• Predicate pushdown — filter pushed into reader.
• Projection pushdown — select pushed into reader.
• .explain() — inspect the plan.
• .explain(streaming=True) — streaming physical plan.
• over() — SQL-style window; per-partition aggregate/rank.
• rolling_mean(window_size=N) — row-based rolling.
• rolling_mean_by(by="ts", window_size="7d") — time-based rolling.
• cum_sum / cum_max / cum_min — cumulative.
• rank / dense_rank / ordinal — ranking.
• shift(N) — lag/lead.
• Join in-memory — small right side; hash join.
• External merge join — both large; sort + merge on disk.
• Partitioned group-by — hash-partition, spill per bucket.
• POLARS_TMPDIR — spill directory override.
• DuckDB for complex SQL — better multi-way join optimiser.
• Interop via Arrow — zero-copy between Polars/DuckDB/Pandas.
• Pandas for < 10 GB — interactive analysis.
• Polars for TB-scale — batch ETL.
• .write_parquet — materialised write (not streaming).
• .sink_parquet — streaming write (out-of-core).

Frequently asked questions

When should I use collect(streaming=True) vs .collect()?

Use collect(streaming=True) when the data does not fit in RAM (larger than ~50% of available memory) — the streaming engine reads chunks and spills hash partitions or sort buffers to disk when needed. Use .collect() (default) when data fits in RAM — no streaming overhead, faster execution. Both compile from the same lazy plan; the terminal call is the only difference. Rule of thumb — start with .collect() for exploration; switch to .collect(streaming=True) or .sink_parquet() when moving to production with real-scale data.

Does streaming work for all operations?

Most common operations are streaming-eligible: filter, project, join (hash or external merge), group-by (partitioned hash spill), sort (external merge sort), aggregation, cumulative functions, most window functions via over(), rolling functions. Some complex window compositions and deeply nested expressions fall back to non-streaming; verify with lf.explain(streaming=True) — the plan will indicate STREAMING: at the top if the whole pipeline streams, or show fallback nodes. When streaming falls back, either restructure the pipeline (split into stages), or drop to DuckDB SQL for the problem step.

How does Polars compare to DuckDB?

Both are Rust-native, both stream, both work with Arrow / Parquet. Polars has a DataFrame API preferred by Python engineers and expression fusion; DuckDB has SQL and a stronger cost-based optimiser for complex multi-way joins and GROUPING SETS. In practice teams use both — Polars for the expression-fluent ETL steps and DuckDB for the complex SQL steps — with zero-copy Arrow interop between them. For pure benchmark on simple ETL, Polars often wins by 20-40% on streaming workloads; DuckDB often wins by 20-40% on complex join workloads. Rule of thumb — Polars for pipeline expressions; DuckDB for ad-hoc SQL analysis; both can share Parquet files without copy.

What's expression fusion and why does it matter?

Polars compiles chained expressions into a single lower-level Rust operation when possible — e.g., filter().with_columns().group_by() becomes one pass instead of three. Critical for streaming performance because each additional pass would mean re-reading the data. Fusion happens at plan compilation; the more chaining you do in a single lazy pipeline (without breaking with .collect()), the better the fusion. Rule — one continuous lazy chain; avoid intermediate .collect() calls unless required.

How do I handle CSVs with mixed schemas?

Use scan_csv(path, infer_schema_length=10000, ignore_errors=True) — infer schema from more rows (default 100) and skip malformed lines. For dirty CSVs, cast columns explicitly with .with_columns([pl.col("x").cast(pl.Float64, strict=False)]) — strict=False makes cast return null instead of raising on parse failure. For heavily mixed schemas, read as all-string and parse per-column with .str.to_integer() / .str.to_date() etc. Rule — always inspect a sample first with pl.read_csv(path, n_rows=100) before designing the pipeline.

Should I still use pandas for anything?

Yes — Pandas is best for interactive analysis on smaller data (< 10 GB), for ecosystem integration (matplotlib, sklearn, statsmodels compat), and for legacy code you're maintaining. Polars has better performance and memory characteristics for pipeline code, and DuckDB has better SQL. Rule — Pandas for exploration and small-data plotting; Polars for pipelines; DuckDB for SQL. They interop cleanly via Arrow / Parquet — you don't have to pick one.

Practice on PipeCode

• Drill the SQL practice library → — 450+ DE-focused questions covering DataFrame patterns.
• Sharpen SQL aggregation drills → for group-by patterns.
• Layer SQL join drills → — Polars joins mirror SQL semantics.
• Warm up with SQL optimization drills → — understand how streaming saves resources.
• Layer window function drills → — Polars over() mirrors SQL windows.
• For the broader SQL interview surface, take the SQL for Data Engineering course →.

Pipecode.ai is Leetcode for Data Engineering — every `polars streaming` pattern above ships with hands-on practice rooms where you convert a 500 GB Parquet aggregation from Pandas OOM to Polars `sink_parquet` success on a laptop, choose between eager / lazy / streaming based on data-to-RAM ratio, verify predicate + projection pushdown via `.explain()`, compose rolling windows over sensor time-series, and finally decide between Polars DataFrame API and DuckDB SQL based on join complexity and team fluency — the exact local-TB-scale ETL fluency that senior DE interviews probe.

Practice SQL now →
Aggregation drills →