Polars 惰性管道实战：200 万行日志从 scan_csv 到流式聚合

日志分析这件事，用 pandas 做十万行很舒服，做一百万行开始卡，做到几百万行就得开始琢磨 dtype 优化、分块读取、还有那个永远算不准的内存占用。真正让人难受的不是慢，是不可预测——同一个脚本，换个月份的文件就可能把机器内存吃穿。

这两年 Polars 在数据处理圈子里热度一直很高，1.x 版本之后 API 也基本稳定了。它不是”更快的 pandas”，底座就不一样：列式存储、Rust 实现、多线程执行、以及一套真正可用的惰性查询计划。这篇文章不走”Polars 和 pandas 谁更快”的老路，而是直接开一个完整项目，从生成数据一路跑到产出报表，把中间那些会卡住你的地方说清楚。

环境只需要一个包：pip install "polars>=1.0"。以下代码在 Polars 1.x 上验证过，更老的版本有些 API 名字不同，后面会单独提。

一、先造一份有问题的数据

不要用干净的示例数据写教程，干净数据会掩盖绝大多数坑。这里用标准库生成 200 万行，字段故意混着类型问题：时间戳是字符串、状态码是整数但有空值风险、耗时字段偶尔出现异常大的值。

import csv
import random
import datetime

random.seed(7)

LEVELS = ["INFO", "INFO", "INFO", "DEBUG", "WARN", "ERROR"]
MODULES = ["auth", "billing", "search", "upload", "notify"]
STATUS = [200, 200, 200, 200, 200, 404, 502]
ROWS = 2_000_000

def build(path, rows=ROWS):
    base = datetime.datetime(2025, 3, 1, 0, 0, 0)
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["ts", "level", "module", "status", "ms", "uid"])
        for i in range(rows):
            stamp = base + datetime.timedelta(seconds=i * 0.05)
            writer.writerow([
                stamp.strftime("%Y-%m-%d %H:%M:%S"),
                random.choice(LEVELS),
                random.choice(MODULES),
                random.choice(STATUS),
                random.randint(3, 1800),
                random.randint(1, 5000),
            ])

if __name__ == "__main__":
    build("app.log.csv")

跑完会得到一个一百多兆的 CSV。这就是我们接下来所有操作的对象。

二、第一步就不该是 read_csv

大多数人的肌肉记忆是 pl.read_csv()。在这个场景下它并不合适——read_csv 会立刻把整份数据读进内存，和 pandas 一样，只是快一些。我们真正要的是 scan_csv：它不读数据，只读表头和一小段样本做类型推断，返回的是一个 LazyFrame，也就是一张”待执行的查询计划”。

import polars as pl

lf = pl.scan_csv(
    "app.log.csv",
    schema_overrides={
        "status": pl.Int16,
        "uid": pl.Int32,
    },
)

schema_overrides 这个参数值得多说两句。uid 的最大值是 5000，用 Int32 就够，没必要用默认的 Int64。在两百多万行的规模下，把一个 Int64 列降成 Int16，节省的内存是实打实的，而这类改动在惰性管道里几乎没有成本。如果你的列是固定枚举值（像上面的 level 和 module），还可以直接用 pl.Enum 类型，字符串比较会快很多。

此时执行输出看看计划：

print(lf.explain())

你会看到一棵简单的树，只有”扫描文件”一个节点。这说明数据还没被读，好戏在后面。

三、清洗：全部用表达式，别碰 Python 函数

Polars 最重要的心智模型是：你写的不是操作，是表达式。表达式会被优化器统一处理、向量化执行、多线程并行。一旦你在里面塞了 map_elements 或者绕过表达式直接写 Python 循环，就等于主动放弃了它所有的优势。

clean = (
    lf
    .filter(pl.col("level").is_in(["WARN", "ERROR"]))
    .with_columns(
        pl.col("ts").str.to_datetime("%Y-%m-%d %H:%M:%S", strict=False),
        pl.col("ms").clip(0, 5000),
        (pl.col("status") >= 500).alias("is_5xx"),
    )
)

几个细节：

• with_columns 对应 pandas 的 assign，但它接受多个表达式，一次性算出所有新列，而不是链式地一遍遍复制表。每多一次 with_columns，优化器就多一层节点，虽然会自动合并，但能一次写完就别拆。
• str.to_datetime 的 strict=False 是把脏数据兜住。默认情况下一旦有解析不了的字符串会直接抛异常，在真实日志里这几乎必然发生。
• clip 用来压掉脏数据里的异常耗时。用表达式做这件事，全程在 Rust 里跑；如果换成 apply 一个 Python lambda，两百多万行会慢上两个数量级。

还有一个坑必须提前说：在 filter 里，不能用 and / or / not。Python 的 and 会试图把两个表达式对象当成布尔值求值，结果直接报错或者给出诡异结果。必须换成 &、|、~，而且每个条件都要加括号——位运算符的优先级比比较运算符高，不加括号就是错的。

# 正确
.filter((pl.col("status") == 200) & (pl.col("ms") > 900))

# 错误：运算符优先级导致条件失效
.filter(pl.col("status") == 200 & pl.col("ms") > 900)

四、聚合：一次算完所有指标

summary = (
    clean
    .group_by("module")
    .agg(
        pl.len().alias("err_cnt"),
        pl.col("ms").mean().round(1).alias("avg_ms"),
        pl.col("ms").quantile(0.95).alias("p95_ms"),
        pl.col("uid").n_unique().alias("affected_users"),
        pl.col("is_5xx").sum().alias("s5xx_cnt"),
    )
    .sort("p95_ms", descending=True)
)

pl.len() 是统计组内行数。Polars 1.x 之前的写法是 pl.count()，现在会被废弃警告，新代码直接写 pl.len()。

注意这里仍然没有真正计算——summary 还是一个 LazyFrame。所有过滤、类型转换、分组聚合都被记在计划里，等待一个触发点。这个设计的好处是优化器能看到全貌：比如它发现我们只需要 module、ms、uid、is_5xx 这几列，就会在扫描阶段做列裁剪，把 ts 那列直接跳过不读。CSV 解析的时间省下来是相当可观的。

你可以用 summary.explain() 看到这个裁剪发生了：箭头图里 scan_csv 节点后面会标注 PROJECT 4/6 COLUMNS 之类的信息。

五、加一张维表

报表里只有 module 名字没用，业务方要看的是负责人。造一个小维表然后 join：

owners = pl.DataFrame({
    "module": ["auth", "billing", "search", "upload", "notify"],
    "owner": ["team-a", "team-a", "team-b", "team-b", "team-c"],
})

final = summary.join(owners, on="module", how="left")

这里 owners 是小表，join 会被优化器处理成广播。如果你 join 的两边都很大，记得确认一下 on 的列类型一致——Polars 对类型匹配比 pandas 严格，Int32 和 Int64 在某些版本上 join 会给出预期外的结果，直接 cast 成一致最省心。

六、触发执行：普通 collect 和流式 collect

到这里才第一次真正读文件。最直接的方式：

result = final.collect()
print(result)

但如果数据规模再涨一个数量级，内存又会开始紧张。这时候用流式执行引擎，它会分块扫描、分块聚合，内存占用基本恒定：

result = final.collect(engine="streaming")

这里有个版本坑：Polars 1.0 之前是 collect(streaming=True)，1.0 之后改成了 engine="streaming"。如果你在旧版上照抄新写法会报参数错误，反过来新版用 streaming=True 也会得到废弃提示。升级版本时特别容易撞上。

顺带说一句，聚合这类操作用流式引擎收益最大；但涉及需要全量排序的操作（比如 sort 一整张没聚合的大表），流式引擎不一定能省多少内存，排序本身就是需要缓冲的。别把 engine="streaming" 当成万能内存开关。

七、把 5xx 错误单独落成 Parquet

分析完还要给下游用，这时候 sink_parquet 比 write_parquet 更合适——前者边算边写，不需要先在内存里拼出完整结果：

(
    clean
    .filter(pl.col("is_5xx"))
    .select("ts", "module", "status", "ms", "uid")
    .sink_parquet("errors_5xx.parquet")
)

生成的 Parquet 文件会小得让你意外。原始 CSV 一百多兆，过滤后剩几十万行，存成 Parquet 通常只有几兆，而且列裁剪、谓词下推这些优化在后续读取时都能继续生效。这也是为什么稍微正经一点的数据管道，中间层基本都用 Parquet 而不是 CSV。

八、pandas 迁移踩坑清单

如果你手上有一堆 pandas 代码想搬过来，下面这几条基本都会撞上：

• 没有行索引。Polars 完全没有 index 概念，df.loc[]、.iloc[] 一律不存在。要取某一行用 df.slice(offset, length) 或 df.row(n)。
• groupby 变成 group_by。下划线是老 API 的残留，新代码统一用 group_by。
• 不是原地修改。with_columns、filter 都返回新对象，调完必须接住返回值。df.with_columns(...) 单独写一行什么都不会发生。
• 列访问用 pl.col("x")，不要写 df.x 或者 df["x"]。后者在 Polars 里是取列的另一种语法，但会退化成非惰性操作，和表达式体系是两套东西。
• 空值和 NaN 是两回事。Polars 里 null 是缺失，NaN 是浮点特殊值，判断要分别用 is_null() 和 is_nan()，别指望一个 isna() 通吃。
• 别用 apply 做能写成表达式的事。map_elements 是逐元素调用 Python，会把多线程优势全部抵消，只在实在无法向量化时才用。

九、这份数据不适合用 Polars 的情况

写完上面这些，很容易得出”以后都用 Polars”的结论。实际上不是：

• 数据只有几千行的时候，pandas 的交互式体验依然更好，Polars 的表达式语法在小数据上反而是负担。
• 下游生态强绑 pandas 的场景——sklearn 的 fit、statsmodels、大部分绘图库，都还是 pandas 更省事。这时候用 df.to_pandas() 转过去就行，别硬扛。
• 需要很复杂的行级业务逻辑，一行里要根据十几个字段做分支判断。这种逻辑用 map_elements 写出来在 Polars 里既不快也不好看，不如老老实实写 Python 循环。

判断标准其实很简单：如果你是”读一堆文件、筛筛洗洗、算几个聚合、写回去”，Polars 的惰性管道几乎是为你量身定做的；如果是”小数据、交互式探索、和一堆 Python 库来回倒腾”，pandas 仍然是默认答案。

完整跑一遍上面的代码，从生成数据到输出 Parquet，在一台普通笔记本上大概十几秒。真正省下的不只是时间，是那种”数据量涨十倍，脚本要不要重写”的焦虑。