Build Polars capability by building a pipeline

This is a web presentation version of the tutorial. It is not a syntax tour. It is a guided build: generate data, scan lazily, transform with expressions, validate joins, test the output, and sink a production-shaped report.

Think query engine, not spreadsheet

The default pipeline shape

Use scan_* plus chained expressions, then call collect() once at the edge — or use sink_parquet() when writing output.

The 12 patterns you need first

What you will build

polars-workalong/
├── data/
│   ├── raw/
│   └── output/
├── src/polars_workalong/
│   ├── generate_data.py
│   └── pipeline.py
├── tests/test_pipeline.py
├── pyproject.toml
└── POLARS_WORKALONG_TUTORIAL.md

Start from a clean environment

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -e '.[dev]'

python - <<'PY'
import polars as pl
print(pl.__version__)
print(pl.show_versions())
PY

Do not skip environment verification. Version drift creates fake learning problems.

Generate realistic input data

python -m polars_workalong.generate_data

The point is not fake demo data. The point is a controlled fixture that forces joins, nulls, dates, strings, and nested values.

DataFrame, LazyFrame, Series, expressions

sales = pl.DataFrame({
    "order_id": [1, 2, 3, 4],
    "region": ["north", "south", "north", "west"],
    "amount": [100.0, 75.5, 210.0, 40.0],
    "is_returned": [False, False, True, False],
})

result = sales.select(
    "order_id",
    (pl.col("amount") * 1.1).round(2).alias("amount_plus_10pct"),
)

Replace row logic with expression logic

answer = sales.with_columns(
    pl.when(pl.col("is_returned"))
      .then(0.0)
      .otherwise(pl.col("amount"))
      .alias("net_amount"),

    pl.when(pl.col("amount") < 75).then(pl.lit("small"))
      .when(pl.col("amount") < 175).then(pl.lit("medium"))
      .otherwise(pl.lit("large"))
      .alias("amount_band"),
)

assert answer["net_amount"].to_list() == [100.0, 75.5, 0.0, 40.0]
assert answer["amount_band"].to_list() == ["medium", "medium", "large", "small"]

Schemas are contracts

customers = pl.scan_parquet("data/raw/customers.parquet")
events = pl.scan_parquet("data/raw/events.parquet")

print(customers.collect_schema())
print(events.collect_schema())

Types are not decoration. They decide what operations are legal, how values are parsed, whether joins work, and how memory is represented.

Lazy scanning and optimizer pushdown

query = (
    pl.scan_parquet("data/raw/events.parquet")
      .select("user_id", "event_date", "event_type", "amount")
      .filter(pl.col("event_date") >= pl.date(2026, 1, 1))
      .filter(pl.col("amount") > 0)
)

print(query.explain())

If you select only four columns and filter early, Polars can avoid reading and processing work you never needed.

Do not collect in the middle

df = pl.scan_parquet(path).collect()
df = df.select(...)
df = df.filter(...)

df = (
    pl.scan_parquet(path)
      .select(...)
      .filter(...)
      .collect()
)

select vs with_columns

events.select(
    "user_id",
    "amount",
    (pl.col("amount") * 1.0825).alias("with_tax"),
)

events.with_columns(
    pl.col("device").fill_null("unknown"),
    pl.col("event_date").dt.truncate("1mo").alias("month"),
)

Filtering correctly

filtered = events.filter(
    (pl.col("event_type") == "purchase")
    & (pl.col("amount") > 0)
    & (pl.col("event_date").is_between(pl.date(2026, 1, 1), pl.date(2026, 6, 30)))
)

Null and NaN are not the same

clean = df.with_columns(
    pl.col("device").fill_null("unknown"),
    pl.col("amount").fill_null(0.0).fill_nan(0.0),
)

Do not pretend missing data and invalid numeric output are the same problem.

Strings, dates, and time buckets

events.with_columns(
    pl.col("event_type").str.to_lowercase().alias("event_type_clean"),
    pl.col("event_date").dt.truncate("1mo").alias("month"),
    pl.col("event_date").dt.weekday().alias("weekday"),
)

String and date work should stay inside expressions. Do not parse dates or clean strings row-by-row in Python unless you have no real alternative.

Group and aggregate without losing meaning

monthly = (
    events
      .group_by("month", "region", "plan")
      .agg(
          pl.len().alias("events"),
          pl.col("user_id").n_unique().alias("users"),
          pl.col("net_amount").sum().round(2).alias("net_revenue"),
      )
      .with_columns(
          (pl.col("net_revenue") / pl.col("users")).round(2).alias("revenue_per_user")
      )
)

Join with cardinality validation

events.join(
    customers,
    on="user_id",
    how="inner",
    validate="m:1",
)

A join is not just syntax. It is a claim about the relationship between tables.

Windows answer “within each group” questions

ranked = events.with_columns(
    pl.col("amount")
      .rank("dense", descending=True)
      .over("user_id")
      .alias("purchase_rank_for_user")
)

A group-by collapses rows. A window expression keeps rows and adds group-aware calculations.

Duplicate policy, nested data, reshaping

events.select(
    "event_id",
    pl.col("tags").list.first().alias("primary_tag"),
)

Concatenation, plans, profiling, streaming

query = build_event_report(events_path, customers_path)
print(query.explain())

# In-memory result
report = query.collect()

# Disk output without forcing an intermediate DataFrame
query.sink_parquet("data/output/monthly_region_plan_report.parquet")

Use collect() when you need a DataFrame in memory. Use a sink when the real output is a file.

Why map_elements is usually the wrong move

df.with_columns(
    pl.col("amount").map_elements(lambda x: x * 1.0825)
)

df.with_columns(
    (pl.col("amount") * 1.0825).alias("amount_with_tax")
)

Test transformations like business logic

from polars.testing import assert_frame_equal

actual = build_event_report(events_path, customers_path).collect().sort("region")
expected = pl.DataFrame({...}).sort("region")

assert_frame_equal(actual, expected)

SQL interface and ecosystem boundaries

ctx = pl.SQLContext()
ctx.register("events", events)
result = ctx.execute("""
    SELECT event_type, COUNT(*) AS events
    FROM events
    GROUP BY event_type
    ORDER BY events DESC
""").collect()

Use SQL when it makes the transformation clearer. Use Polars expressions when you need composable, typed Python pipelines. Convert to pandas only at the boundary when another library requires it.

Monthly region and plan report

def build_event_report(events_path: Path, customers_path: Path) -> pl.LazyFrame:
    events = (
        pl.scan_parquet(events_path)
          .select("event_id", "user_id", "event_date", "event_type", "amount", "device")
          .filter(pl.col("event_date").is_between(pl.date(2026, 1, 1), pl.date(2026, 6, 30)))
          .with_columns(
              pl.col("device").fill_null("unknown"),
              pl.when(pl.col("event_type").is_in(["purchase", "refund"]))
                .then(pl.col("amount"))
                .otherwise(0.0)
                .fill_null(0.0)
                .alias("net_amount"),
          )
          .with_columns(
              (pl.col("net_amount") * 1.0825).round(2).alias("amount_with_tax"),
              pl.col("event_date").dt.truncate("1mo").alias("month"),
          )
    )

Join, aggregate, derive, sort

customers = (
    pl.scan_parquet(customers_path)
      .select("user_id", "region", "plan", "is_active")
      .filter(pl.col("is_active"))
)

return (
    events.join(customers, on="user_id", how="inner", validate="m:1")
      .group_by("month", "region", "plan")
      .agg(
          pl.len().alias("events"),
          pl.col("user_id").n_unique().alias("users"),
          pl.col("net_amount").sum().round(2).alias("net_revenue"),
          pl.col("amount_with_tax").sum().round(2).alias("revenue_with_tax"),
      )
      .with_columns((pl.col("net_revenue") / pl.col("users")).round(2).alias("revenue_per_user"))
      .sort("month", "revenue_with_tax", descending=[False, True])
)

Done means tested, not “looks right”

python -m polars_workalong.generate_data
python -m polars_workalong.pipeline
python -m pytest -q

Deliberate practice sequence

1. Rewrite one pandas transformation as Polars expressions.
2. Add one schema boundary and one tiny known-output test.
3. Inspect the lazy plan before and after moving select and filter.
4. Add a join with validate=... and intentionally break the cardinality.
5. Sink output to Parquet and read it back with scan_parquet.

The goal is not to know every method. The goal is to build reliable pipelines without falling back to row loops.

What capability looks like