上个月帮一个做电商的朋友看他们的对账脚本。脚本是两年前写的，pandas 一把梭，数据量涨到 2000 多万行以后，跑一次要二十多分钟，32G 内存的机器经常在中途被 OOM Killer 干掉。脚本本身逻辑并不复杂：读 CSV、剔掉退款和测试单、按门店和月份汇总、再算环比和排名。真正拖后腿的是写法——pd.read_csv 先把整张表塞进内存，然后用两个 apply 分别做时间拆解和渠道映射，最后一个 for 循环逐店算环比。
我用 Polars 重写了一版。同样的数据，同样的机器，全流程 50 秒左右，峰值内存没超过 3GB。下面把整个过程拆开讲，代码可以直接跑，中间几个坑我也会标出来。
一、先搞清楚慢在哪
原脚本有三个地方在持续吃内存：
read_csv读进来以后，字符串列全部是 Pythonobject，2000 万个 Python 字符串对象光是对象头就占掉一大半内存；apply会把每一列变成 Python 层的逐行调用，等于把向量化计算退化成了 for 循环；- 中间结果全都在内存里落地，没有任何一步是流式处理的。
换成 Polars 之后思路是反过来的：不读，先 scan；不循环，用 表达式链；不落中间表，直接 流式写出。这三件事是 Polars 提速的主要来源，跟“它用 Rust 写的”关系没那么大。
二、装环境和造一份能复现的数据
依赖只有两个：
pip install polars numpy如果你用 uv，对应的是 uv pip install polars numpy。版本上我这边是 Polars 1.x，1.0 之后有几个 API 变了：pl.count() 现在叫 pl.len()，collect(streaming=True) 在新版本里推荐写成 collect(engine="streaming")。本文代码按 1.x 写，老版本稍微改一下也能跑。
用 numpy 造 2000 万行订单，字段贴近真实业务：订单号、门店、用户、渠道、金额、成本、时间戳。
import numpy as np
import polars as pl
n = 20_000_000
rng = np.random.default_rng(20240607)
df = pl.DataFrame({
"order_id": np.arange(1, n + 1, dtype=np.int64),
"shop_id": rng.integers(1, 501, n).astype(np.int32),
"user_id": rng.integers(1, 2_000_001, n).astype(np.int32),
"channel": rng.choice(["app", "mini", "web", "h5"], n),
"amount": np.round(rng.gamma(2.0, 50.0, n), 2),
"cost": np.round(rng.gamma(2.0, 35.0, n), 2),
"ts": np.datetime64("2024-01-01")
+ rng.integers(0, 366 * 24 * 3600, n).astype("timedelta64[s]"),
})
df.write_csv("orders.csv")
print(df.estimated_size("mb")) # 大约 1.1 GB跑完会得到一个 1GB 出头的 CSV。如果你机器内存只有 16G，把 n 改成 500 万也能对照着看效果，后面的代码不用动。
三、第一步：不要 read，要 scan
这是最容易改也最容易被忽略的一步。pl.scan_csv 返回的是 LazyFrame，它只是记下“我要读这个文件”这件事，并不真的读。后续所有操作都会拼成一个逻辑计划，直到你调用 collect() 才真正执行。
import polars as pl
lf = pl.scan_csv("orders.csv", try_parse_dates=True)
print(lf.explain())explain() 会打印出优化后的执行计划，你能看到谓词下推、投影下推这些优化有没有生效。养成在 collect 之前瞄一眼 explain() 的习惯，比事后猜为什么慢要省事得多。
把清洗逻辑接上去。这里顺手做了三件事：过滤掉非正金额、算毛利、从时间戳派生出日期和月份。
lf = (
pl.scan_csv("orders.csv", try_parse_dates=True)
.filter(pl.col("amount") > 0)
.with_columns(
profit = pl.col("amount") - pl.col("cost"),
date = pl.col("ts").dt.date(),
month = pl.col("ts").dt.truncate("1mo"),
)
)注意这里没有任何数据被读进内存。filter 和 with_columns 只是往计划里加节点。
四、第二步：干掉 apply，全部换成表达式
原脚本里有一段鬼东西，大概长这样：
df["channel_name"] = df["channel"].apply(
lambda x: {"app": "自有 App", "mini": "小程序"}.get(x, "其他")
)
df["amount_level"] = df["amount"].apply(
lambda x: "大额" if x > 500 else ("中额" if x > 100 else "小额")
)这种写法在 Polars 里对应的是 pl.when().then().otherwise()，而且是可以直接在 LazyFrame 上跑的表达式，不用 collect：
lf = lf.with_columns(
channel_name = (
pl.when(pl.col("channel") == "app").then(pl.lit("自有 App"))
.when(pl.col("channel") == "mini").then(pl.lit("小程序"))
.when(pl.col("channel") == "web").then(pl.lit("网站"))
.otherwise(pl.lit("其他"))
),
amount_level = (
pl.when(pl.col("amount") > 500).then(pl.lit("大额"))
.when(pl.col("amount") > 100).then(pl.lit("中额"))
.otherwise(pl.lit("小额"))
),
)有一点要提醒：.then() 里如果不是列表达式，得用 pl.lit() 包一下。直接写字符串在早期版本会报错，1.x 里虽然部分场景能隐式转换，但统一用 pl.lit 更保险。
五、第三步：窗口函数算环比和排名
这是 Polars 最舒服的地方。原脚本用 for 循环逐店算环比，现在一行 over() 就够。
先按门店和月份做聚合：
monthly = (
lf.group_by(["month", "shop_id"])
.agg(
pl.len().alias("orders"),
pl.col("amount").sum().alias("gmv"),
pl.col("profit").sum().alias("profit"),
pl.col("user_id").n_unique().alias("buyers"),
)
.with_columns(
arpu = pl.col("gmv") / pl.col("buyers"),
margin = pl.col("profit") / pl.col("gmv"),
)
)然后是环比和当月排名。这里的关键是数据得先按 shop_id, month 排好序，shift() 配合 over() 才有意义。
ranked = (
monthly.sort(["shop_id", "month"])
.with_columns(
prev_gmv = pl.col("gmv").shift(1).over("shop_id"),
)
.with_columns(
mom = pl.col("gmv") / pl.col("prev_gmv") - 1,
m_rank = pl.col("gmv").rank("dense", descending=True).over("month"),
)
.drop("prev_gmv")
)over("shop_id") 的语义等同于 SQL 里的 PARTITION BY shop_id。over("month") 则是按月份分区排名。rank 的第一个参数传 "dense" 表示相同 GMV 并列后名次不跳号，"ordinal" 则是稳定排序不并列，看业务口径选。
六、第四步：如果真要看结果，记得流式写
月度汇总表只有几十万行，直接 collect() 没问题：
result = ranked.collect(streaming=True)
print(result.head())
result.write_parquet("shop_month.parquet")但如果输出的是明细表——比如把清洗后的 2000 万行存成 Parquet 供下游用——就绝对不能用 collect() 再写。用 sink_parquet，它会边算边写，内存占用跟数据量基本无关：
(
lf.select(["order_id", "shop_id", "channel_name",
"amount", "profit", "date", "amount_level"])
.sink_parquet("orders_clean.parquet")
)这两个 API 的区别值得记一下：collect 是“把结果拿回内存”，sink 是“把结果推到磁盘”。数据量上到千万级以后，能 sink 就 sink。
七、我实际踩到的几个坑
1. streaming 不是万能的
某些算子（早期版本的 sort、复杂的 over、部分 join）在流式引擎里支持不完整，Polars 会悄悄退回内存引擎执行，你看到的 streaming=True 其实没生效。判断方法是看 explain() 输出里有没有出现 IN-MEMORY 字样。看到就说明这一段是内存执行的，得考虑换写法。
2. schema 推断只采样前 100 行
scan_csv 默认只读前 100 行来推断列类型。如果前 100 行的 amount 恰好都是整数，后 2000 万行里有小数，就会在 collect 的时候抛类型错误。稳妥的做法是显式传 schema：
pl.scan_csv(
"orders.csv",
schema_overrides={
"order_id": pl.Int64,
"shop_id": pl.Int32,
"amount": pl.Float64,
"cost": pl.Float64,
},
try_parse_dates=True,
)3. 字符串列比想象中吃内存
2000 万行如果 channel 只有 4 个取值，用 pl.Categorical 能把这列的内存占用降到原来的几十分之一。转换开销很小，扫描阶段直接指定就行：
pl.scan_csv("orders.csv", schema_overrides={"channel": pl.Categorical})4. 别对 LazyFrame 重复 collect
LazyFrame 是惰性的，每次 collect() 都会从头执行一遍完整计划。如果有人为了调试写了两次 collect()，那就是把 50 秒干成了 100 秒。调试阶段用 .head(1000).collect() 取一小坨看结构，确认没问题再跑全量。
5. Polars 没有索引这个概念
从 pandas 转过来最不适应的一点。所有基于 .loc[]、.iloc[]、.set_index() 的习惯都得扔掉，改成 filter + select。写起来更啰嗦一点，但换来的好处是没有隐式索引对齐带来的各种意外。
八、什么时候别用 Polars
说点实在的。Polars 不是无脑替换 pandas 的银弹，下面几种情况我一般还是留在 pandas 里：
- 要和 sklearn、statsmodels 直接对接。它们只认 DataFrame 和 numpy 数组，来回
to_pandas()转换在千万级数据上是有成本的，除非你在 Polars 里已经把数据压到几十万行； - 数据量小于十万行。这个量级 pandas 和 Polars 的耗时差异在秒级以内，Polars 稍微快一点，但不够抵消重写脚本的成本；
- 重度依赖 pandas 生态里的边角功能。比如多层索引的复杂切片、
resample的某些冷门参数。Polars 能覆盖 90% 的常见场景，剩下 10% 迁过去反而更痛苦。
换个说法：如果脚本一次跑超过 5 分钟，或者内存占用让你不敢在笔记本上开，那值得迁。否则先放着。
九、完整脚本
把上面的东西串起来，这是最终版本，可以直接存成 etl.py 跑：
import time
import polars as pl
CSV_IN = "orders.csv"
PARQUET_OUT = "shop_month.parquet"
t0 = time.perf_counter()
lf = (
pl.scan_csv(
CSV_IN,
schema_overrides={"channel": pl.Categorical},
try_parse_dates=True,
)
.filter(pl.col("amount") > 0)
.with_columns(
profit = pl.col("amount") - pl.col("cost"),
date = pl.col("ts").dt.date(),
month = pl.col("ts").dt.truncate("1mo"),
channel_name = (
pl.when(pl.col("channel") == "app").then(pl.lit("自有 App"))
.when(pl.col("channel") == "mini").then(pl.lit("小程序"))
.when(pl.col("channel") == "web").then(pl.lit("网站"))
.otherwise(pl.lit("其他"))
),
amount_level = (
pl.when(pl.col("amount") > 500).then(pl.lit("大额"))
.when(pl.col("amount") > 100).then(pl.lit("中额"))
.otherwise(pl.lit("小额"))
),
)
)
monthly = (
lf.group_by(["month", "shop_id"])
.agg(
pl.len().alias("orders"),
pl.col("amount").sum().alias("gmv"),
pl.col("profit").sum().alias("profit"),
pl.col("user_id").n_unique().alias("buyers"),
)
.with_columns(
arpu = pl.col("gmv") / pl.col("buyers"),
margin = pl.col("profit") / pl.col("gmv"),
)
)
final = (
monthly.sort(["shop_id", "month"])
.with_columns(prev_gmv = pl.col("gmv").shift(1).over("shop_id"))
.with_columns(
mom = pl.col("gmv") / pl.col("prev_gmv") - 1,
m_rank = pl.col("gmv").rank("dense", descending=True).over("month"),
)
.drop("prev_gmv")
.sort(["month", "m_rank"])
)
result = final.collect(streaming=True)
result.write_parquet(PARQUET_OUT)
print(f"rows = {result.height}")
print(f"peak = {result.estimated_size('mb'):.1f} MB")
print(f"cost = {time.perf_counter() - t0:.1f}s")我这边 16 核 / 32G 的机器上，2000 万行跑完大概 48 到 55 秒，estimated_size 在 40MB 上下（月度汇总表本身不大），过程中系统监控看到的峰值常驻内存大概 2.8G。同一份数据用原来的 pandas 脚本，read_csv 这一步就要四五分钟，中间表叠加起来能把机器吃干。
如果你们的数据量还在往上涨，下一步可以看两个方向：一是把 CSV 换成 Parquet 存档，scan_parquet 比 scan_csv 又要快一截，因为不用解析文本；二是把清洗后的明细直接接到 DuckDB 上做即席查询，Polars 的 LazyFrame 和 DuckDB 可以直接互转，不用落地。这两条我后面再单独写一篇。
