Prototype on a laptop, scale to 16 billion rows: one Polars query

A pattern we often see at data teams: exploration happens on a laptop in one tool, and once the dataset outgrows the laptop, the pipeline is rewritten in a second tool that scales. From that point on the team maintains two implementations of the same logic, and the two tend to drift apart. Every change has to be made twice, and once the implementations drift, the dashboard and the notebook start disagreeing about the same number.

This post shows a workflow without the rewrite. We take a subset of a large dataset, explore it locally, and build an ETL pipeline from what we find. Then we run the exact same Polars queries on the full month of Polymarket orderbook data, 16 billion rows, on Polars Cloud. What changes between the laptop and the cluster is the method call that decides where execution happens. The results feed a Plotly Dash dashboard hosted on Plotly Cloud.

The dashboard the pipeline feeds, live to explore.

The workflow

Our example data comes from Polymarket, a prediction market where users trade on the outcomes of real-world events. It publishes its full orderbook as a public dataset: every price update, for every active market, logged with a millisecond timestamp. One hour of that data is around 30 million rows, and the three hours we prototype on total 97 million, which still runs comfortably on a laptop. One month is 16 billion rows in around 500 GB of compressed Parquet, which requires larger machines.

Raw orderbook events land in S3 as hive-partitioned Parquet. Polars transforms that raw data into the smaller files that back each plot. A subset of a few hours has the same shape as the full month, so one LazyFrame serves both scales: collected locally while prototyping, then run distributed on Polars Cloud against the full month. Its output is a handful of small, pre-aggregated Parquet artifacts in S3. Plotly handles visualisation and serving: a Dash app on Plotly Cloud reads those artifacts to power the dashboard and an MCP endpoint. The two halves meet at the artifacts.

We built this project together with the Plotly team, and the write-up spans two posts. This one covers the compute half: decoding, aggregating, and scaling the data with Polars.

Step 1: explore a subset locally

We scope to the first three hours of the month before writing any pipeline code. That is enough data to work out the schema, the payload format, and the transformations the dashboard needs, with the fast feedback of local execution. It is also deliberately more than one hour: an hour is a single leaf partition, while three hours span multiple partitions, so the subset has the same structure as the full month and the query already handles the multi-file scan it will meet at scale.

import polars as pl

pl.Config.set_engine_affinity("streaming")

storage_options = {"aws_region": "us-east-1"}

lf = pl.scan_parquet(
    [
        f"s3://bucket/raw/orderbook/year=2026/month=03/day=01/hour={h:02d}/*.parquet"
        for h in range(3)
    ],
    storage_options=storage_options,
)
print(lf.collect_schema())

collect_schema resolves column names and dtypes from the file metadata without reading any data, so it is the cheapest first look at an unfamiliar dataset. To see actual values, pull a single row and glimpse it:

lf.head(1).collect().glimpse()

glimpse prints one row per column, so the wide data payload stays readable instead of being squashed into a table cell.

Tip: set streaming affinity once, affect all collects

set_engine_affinity is a global config. Call it once at module load and every .collect() in that process uses the streaming engine, which processes data in chunks and keeps memory bounded on scans larger than RAM. It will become the default engine in an upcoming release.

Each row is a raw event: two timestamps, a market_id, an update_type, and a data column holding a JSON payload. A value_counts shows two event types: price_change, which fires whenever the best bid or ask on a market moves, and the much rarer book_snapshot, which we only use for the dashboard’s depth chart. The price_change payload holds eight JSON keys, of which we need three. Polars decodes them inline with str.json_decode and a pl.Struct schema that declares only those three fields, so the rest is never parsed, and casts the string-quoted prices to Float64 along the way. The side column becomes a pl.Enum, a categorical type with a fixed set of values that is safe to use across distributed workers. We build a single LazyFrame, price_changes, that filters, decodes, and computes the spread:

PRICE_CHANGE_SCHEMA = pl.Struct({
    "side": pl.String,
    "best_bid": pl.String,
    "best_ask": pl.String,
})

price_change_expr = pl.col("data").str.json_decode(PRICE_CHANGE_SCHEMA)
price_changes = (
    lf.filter(pl.col("update_type") == "price_change")
    .select(
        "timestamp_received",
        "market_id",
        price_change_expr.struct.field("side").cast(pl.Enum(["YES", "NO"])),
        price_change_expr.struct.field("best_bid").cast(pl.Float64),
        price_change_expr.struct.field("best_ask").cast(pl.Float64),
    )
    .with_columns(
        (pl.col("best_ask") - pl.col("best_bid")).alias("spread")
    )
)

df = price_changes.collect()

Prediction market prices

Prices are probabilities between 0 and 1. A YES price of 0.72 means the market prices a 72% chance the event resolves YES. Each market side has a best bid and a best ask, and the spread is the gap between them: narrow in active, liquid markets, wide where trading is thin.

A few group-bys settle the design questions. Activity is heavily skewed: a small set of liquid markets accounts for most of the events, while thousands of markets barely trade. Users will want to drill into one liquid market’s history or scan across markets to compare them. Both views filter on market_id and a timestamp range, so those are the two axes the pipeline output has to be cheap to slice along.

Step 2: shape small artifacts for the app

The dashboard is a web app that could serve many concurrent users, and running a full scan of 16 billion raw events on every page load is not viable. So the heavy aggregation runs ahead of time as a batch job on Polars Cloud, and the dashboard reads only its output.

Tip: separate compute from serve

Run the heavy aggregation once, on a schedule or on demand, and write the results as small artifacts. The app reads only those artifacts, so a page load never scans the raw dataset. The cost of the big scan is paid once per refresh instead of once per visitor.

The pipeline runs in two stages. Stage 1 is the decode we built while exploring: it filters, decodes, and casts the raw events, and sinks the typed result to an intermediate prefix on S3. Stage 2 scans those intermediates, so each aggregation job works with typed columns instead of re-parsing JSON strings:

price_changes = pl.scan_parquet(
    "s3://bucket/artifacts/_intermediate/price_changes/",
    storage_options=storage_options,
)

From that one LazyFrame, Stage 2 writes six artifacts, one per dashboard view:

The heaviest artifact is under 5 MB, so clicking through markets never triggers a scan of the raw data. All six jobs are the same shape: filter, group_by, agg, sink_parquet. One is enough to show the pattern. Here is the activity pulse:

activity_pulse = (
    price_changes.with_columns(
        pl.col("timestamp_received").dt.truncate("1m").alias("minute_bucket")
    )
    .group_by("minute_bucket")
    .agg(
        pl.len().alias("events"),
        pl.n_unique("market_id").alias("live_markets"),
    )
    .sort("minute_bucket")
)

activity_pulse.sink_parquet("artifacts/activity_pulse.parquet")

Tip: dt.truncate + group_by instead of group_by_dynamic

group_by_dynamic requires the index column to be sorted, which on unsorted data means a global sort: expensive, and hard to distribute. Truncating a timestamp to a bucket and using a plain group_by produces the same result for non-overlapping tumbling windows, with one caveat: buckets without events are omitted instead of appearing as zero rows. It is fully partitionable: each worker can aggregate its own chunk independently.

Step 3: run the same query on the full dataset

The full dataset covers one month of trading: 16 billion raw events in 500 GB of compressed Parquet. This is too much for a laptop, so we scale up to a cluster. The pipeline code stays the same. The only difference is how the query executes.

import polars_cloud as pc

pc.authenticate()

ctx = pc.ComputeContext(workspace="my-workspace", name="polymarket-pipeline")
ctx.start()

With the context running, moving a query to the cluster is one change to the call:

activity_pulse.remote(ctx).sink_parquet(
    "s3://bucket/artifacts/activity_pulse.parquet",
    sink_to_single_file=True,
    storage_options=storage_options,
)

The LazyFrame is the same object in both calls. remote(ctx) hands it to Polars Cloud, which distributes the scan and aggregation across the cluster. A distributed sink writes its output as multiple files by default, and sink_to_single_file=True merges it into one. The decode we wrote to eyeball three hours of data now processes 16 billion rows, and there is no second implementation to keep in sync with the prototype.

Step 4: hand the artifacts to Plotly Dash

The dashboard reads the manifest on startup and loads the global artifacts (market profile, activity pulse) into memory. Per-market data is fetched on demand. prices.parquet is written with sink_to_single_file=True and sorted by market_id, so filtering to a single market lets the Parquet reader skip row groups that cannot match. Results are cached on the Plotly Cloud instance with @lru_cache so each market is read from S3 at most once:

from functools import lru_cache

@lru_cache(maxsize=128)
def query_price_and_spread_data(market_id: str) -> pl.DataFrame:
    return (
        pl.scan_parquet(artifact_url("prices.parquet"), storage_options=storage_options)
         .filter(pl.col("market_id") == market_id)
         .collect()
    )

Predicate pushdown

Parquet files store row-group statistics (min/max per column). When a query filters on market_id, the reader checks those statistics and skips row groups whose range cannot contain the target value, before reading any data. Sorting the file by market_id before writing concentrates each market’s rows into a few row groups, which maximises the skipping.

Dash callbacks pass the resulting Polars DataFrame directly to Plotly figure constructors:

@app.callback(
    Output("price-chart", "figure"),
    Output("spread-chart", "figure"),
    Input("market-dropdown", "value"),
)
def update_selected_market(market_id):
    df = query_price_and_spread_data(market_id)
    label = market_name(market_id)
    return price_chart_figure(df, label), spread_chart_figure(df, label)

The same work, scaled

Explore a subset locally, with fast feedback, until the decode and the aggregations are right. Run the same LazyFrames on the full dataset with .remote(ctx), on Polars Cloud or on your own cluster. Serve small pre-aggregated artifacts to a Plotly Dash app, so no page load touches the raw data. Polars transforms the data, and Plotly Dash visualizes and serves it, with a handful of small Parquet files in S3 as the interface between them.