大数据分析中工具：Polars的.apply()和Pandas的.apply()看起来一样，为什么在Polars里用了它等于放弃了向量化和查询优化，什么样的场景下才真的需要逐行处理？

一、同名不同命：两种 .apply() 的语义分野

在 Pandas 中，DataFrame.apply(func, axis=1) 的语义非常直白：把每一行打包成 Series，交给 Python 函数，收集返回值。它本身就是"逐行循环"的语法封装，底层没有太多可优化的空间，用户对此也有合理预期。Pandas 的 Series.apply 同理，逐元素调用。

Polars 早期也提供 .apply()，但在 0.19 版本前后，官方将其重命名并拆分为 map_elements（逐元素）、map_batches（按批）、map_groups（按组）。这次改名本身就是一次"语义澄清"：名字变了，是因为语义责任变了。

本文评述：Polars 的改名不是 API 洁癖，而是一次"防误用设计"。当两个库的函数同名时，迁移者会下意识地认为行为等价；而实际上，Pandas 的 .apply() 是"本来就慢的兜底"，Polars 的逐行调用却是"从快车道主动驶入慢车道"。这个心理落差，才是大量迁移踩坑的根源。

1.1 一张表看清差异

Polars 官方用户指南在 "Expressions and contexts" 与 "User-defined functions" 章节中反复强调：优先使用表达式，只有在表达式无法表达时才使用 UDF（用户自定义函数）。这不是性能建议，而是执行模型的硬约束。

二、执行模型解剖：惰性计划、表达式与向量化内核

要理解"为什么等于放弃优化"，必须先理解 Polars 到底在优化什么。Polars 的核心不是"更快的 Pandas"，而是一个基于 Apache Arrow 内存格式的查询引擎，其设计深受关系代数与列式执行思想影响。

2.1 惰性计划：先画图，再执行

使用 pl.scan_parquet() 或 pl.LazyFrame 时，你写的每一步操作并不会立即执行，而是被记录成一棵逻辑计划树。调用 .collect() 时，优化器才会介入，做谓词下推（predicate pushdown）、投影下推（projection pushdown）、公共子表达式消除、算子融合等。

import polars as pl

lf = (
    pl.scan_parquet("events.parquet")
      .filter(pl.col("dt") >= "2024-01-01")
      .with_columns((pl.col("price") * pl.col("qty")).alias("amount"))
      .group_by("user_id")
      .agg(pl.col("amount").sum())
)
print(lf.explain())   # 查看优化后的计划

执行 explain() 后你会看到，过滤条件被下推到扫描阶段，列裁剪只读取必要字段，聚合在列式内存上批量完成。这些优化的前提是：每个节点都是引擎"看得懂"的表达式。

2.2 表达式系统：声明式而非命令式

Polars 的表达式（pl.col、pl.when、pl.sum 等）是声明式的：你描述"要什么"，引擎决定"怎么算"。表达式可以组合、可以并行、可以在 SIMD 内核上批量执行。这正是 Polars 能在多核上跑满带宽的关键。

而 Python 函数是命令式的：它接收一个标量，返回一个标量，中间可能调用任意库、有副作用、有分支。引擎无法分析它，只能把它当作一个不透明节点。

2.3 向量化内核：Arrow + Rust

Polars 的列式内存基于 Arrow，计算内核用 Rust 编写，能以批（chunk）为单位处理数据，充分利用 CPU 缓存与 SIMD 指令。批量处理意味着循环在编译后的机器码里完成，而不是在 Python 解释器里逐行完成——这两者的开销差距通常是数量级的。

本文评述：很多人把 Polars 的快归因于"Rust 写的"，这只说对了一半。真正决定性能的是"批量 + 列式 + 可优化计划"三者叠加；Rust 只是让这套模型落地得更彻底。反过来，一旦你插入 Python 逐行函数，这三者同时失效，Rust 内核再快也救不了你。

三、为什么逐行调用会切断优化链路

这一节是全文的核心。我们逐条拆解"切断"发生在哪里。

3.1 优化器面对黑盒只能"绕行"

当计划中出现 map_elements 节点时，优化器无法知道这个函数会读取哪些列、是否保持行序、是否可能抛异常、是否依赖外部状态。于是保守策略是：不跨越该节点做下推与融合。原本可以下推到扫描阶段的过滤，可能被迫提前物化；原本可以融合的两个算子，被这个节点隔开。

3.2 逐元素调用 = Python 解释器开销 × 行数

即使函数体只有一行 x * 2，每次调用也要经历：从 Arrow 数组取出标量、构造 Python 对象、进入函数、返回、再装箱回 Arrow。百万行就是百万次这样的往返。这就是所谓"解释器税"。

3.3 并行粒度变粗、负载不均

Polars 的并行通常在算子级别（按列、按块、按分区）。逐行 UDF 节点内部虽然也可能被切块并行，但每个块都要反复进出 Python，且 Python 的 GIL 会限制真正的并行收益。结果是：多核利用率下降，尾延迟上升。

3.4 内存峰值上升

为了把数据交给 Python，引擎往往需要把列式数据转成行式或对象数组，产生额外拷贝。原本流式（streaming）执行可以控制内存，遇到 UDF 节点后可能被迫全量物化。

一句话总结：逐行调用不是"稍微慢一点"，而是把 Polars 赖以生存的三根支柱——可优化计划、向量化内核、细粒度并行——同时抽掉。

四、实测：逐行 vs 向量化的性能落差

为避免空谈，下面给出一组可复现的基准测试方案（读者可在本机复跑）。数据为模拟生成，用于说明量级差异，不代表任何特定硬件结论。

import polars as pl, time, numpy as np

n = 5_000_000
df = pl.DataFrame({
    "a": np.random.rand(n),
    "b": np.random.rand(n),
})

# 方式一：原生表达式（向量化）
t0 = time.perf_counter()
r1 = df.with_columns((pl.col("a") * pl.col("b") + 1).alias("c"))
t1 = time.perf_counter()

# 方式二：map_elements 逐行
t2 = time.perf_counter()
r2 = df.with_columns(
    pl.struct(["a", "b"]).map_elements(
        lambda s: s["a"] * s["b"] + 1, return_dtype=pl.Float64
    ).alias("c")
)
t3 = time.perf_counter()

print(f"expression: {t1-t0:.4f}s")
print(f"map_elements: {t3-t2:.4f}s")

在典型开发机（8 核、16GB 内存）上，500 万行的简单算术，表达式方式通常在几十毫秒量级，而 map_elements 往往在数秒到十几秒量级，差距可达两个数量级。具体数值随硬件、函数复杂度、数据分布变化，但"数量级差距"这一结论在社区基准与官方文档中反复出现。

注：上表倍数为量级参考，来源于社区公开基准与本文模拟测试的整合，非单一实验精确值。读者应以自身数据复测为准。

本文评述：基准测试最容易犯的错，是只测"简单函数"。真实业务里 UDF 往往更复杂（正则、JSON 解析、调用模型），此时 Python 开销占比下降，但优化器失效与内存拷贝的代价依然存在。所以不要用"我的函数很重，所以 UDF 无所谓"来自我安慰——先问一句：这个函数能不能批量化？

五、替代路径全景：从表达式到插件

逐行处理之前，先按下面的"降级阶梯"逐级尝试。绝大多数需求在前三级就能解决。

5.1 第一级：原生表达式

条件分支用 pl.when().then().otherwise()，字符串用 .str 命名空间，时间用 .dt，列表用 .list。Polars 的表达式覆盖面已经相当广。

# 用表达式替代 if/else 逐行判断
df.with_columns(
    pl.when(pl.col("score") >= 90).then(pl.lit("A"))
      .when(pl.col("score") >= 80).then(pl.lit("B"))
      .otherwise(pl.lit("C")).alias("grade")
)

5.2 第二级：map_batches（按批）

当逻辑必须用 Python/NumPy 写，但可以整批处理时，用 map_batches。它把一列（或一个 Series）整体交给函数，函数内部用 NumPy 向量化，开销被摊薄到批次级别。

import numpy as np

def batch_norm(s: pl.Series) -> pl.Series:
    arr = s.to_numpy()
    return pl.Series((arr - arr.mean()) / (arr.std() + 1e-9))

df.with_columns(pl.col("x").map_batches(batch_norm))

5.3 第三级：map_groups（按组）

需要"每组一个模型/一个状态机"时，map_groups 比逐行合理得多：组内仍可向量化，组间可并行。

5.4 第四级：表达式插件（Expression Plugin）

Polars 支持用 Rust 编写自定义表达式插件，注册后即可在表达式中使用，享受完整的优化与并行。这是"既要自定义、又要性能"的终极方案，但门槛较高，适合高频复用的核心逻辑。

5.5 第五级：map_elements（最后手段）

只有在前面都不成立时才用它，并且务必：显式指定 return_dtype、尽量缩小输入列（用 pl.struct 只传必要字段）、先过滤再调用、在文档里注明"性能敏感区"。

拓展阅读：Polars 官方用户指南的 "User-defined functions" 页面、Polars GitHub 的 benchmarks 目录，以及社区教程 docs.pola.rs/user-guide 对各级替代方案有完整示例。

六、真正需要逐行处理的场景清单

逐行不是原罪，滥用才是。以下场景中，逐行（或近似逐行）是合理甚至必要的：

1. 有状态的外部调用：每行要调用一次远程 API、数据库或硬件接口，且无法批量。此时瓶颈在网络，Python 开销可忽略。
2. 复杂解析器：非结构化文本、嵌套 JSON、领域专用格式，现成表达式无法表达，且解析逻辑本身是逐条的。
3. 依赖行内多字段的复杂业务规则：规则数量多、分支深、且频繁变更，用表达式写会变成不可维护的 when 长链。
4. 调用只接受标量的第三方库：某些科学计算、加密、地理编码库没有向量化接口。
5. 原型验证阶段：先跑通逻辑，再逐步下沉为表达式或插件。

本文评述：注意第 1 条与第 4 条的区别——前者瓶颈在外部，逐行是"被迫合理"；后者瓶颈在库的接口设计，逐行是"暂时合理"。前者可以长期保留，后者应列入技术债清单，等待库升级或自研批量化封装。

七、工程迁移路径与检查清单

7.1 迁移四步法

第一步：盘点。全局搜索 .apply(、map_elements，按"能否批量化"分类。

第二步：改写。能表达式化的立即改；能批量的改 map_batches；必须逐行的加注释与告警。

第三步：验证。用 explain() 对比优化前后计划，用基准脚本量化耗时与内存。

第四步：固化。把"禁止无注释的 map_elements"写进代码规范与 CI 检查。

7.2 检查清单

八、前沿追踪与学术预判

近三年，围绕"声明式数据框 + 可扩展 UDF"的研究明显升温。一方面，Polars、DuckDB 等引擎在推动"表达式优先"的编程范式；另一方面，学术界在探索如何在保持优化的前提下安全地嵌入用户代码，例如通过轻量级中间表示、类型推断与副作用分析来"部分理解"UDF。这类工作若能落地，未来 UDF 的"黑盒惩罚"有望降低，但短期内，"能表达式化就表达式化"仍是最稳妥的工程策略。

本文评述：引擎能力的演进不会改变一条基本规律——任何让优化器"看不懂"的代码，都要付出代价。与其等待引擎变聪明，不如先把数据管道的"可分析性"当成一等公民来设计。

九、结语：把边界写进代码规范

Polars 的 .apply() 与 Pandas 的 .apply() 看起来一样，是因为它们共享了"逐行调用 Python 函数"的表层语义；但它们所处的执行环境截然不同：一个本就没有优化器，一个是以优化器为生命线的引擎。在 Polars 里随手写 .apply()，等于在高速公路上突然下车步行——不是不能走，而是你放弃了这条路的全部意义。

把"表达式优先、批量次之、逐行最后"写成团队规范，把 explain() 与基准脚本纳入 CI，比记住任何单个 API 都更有长期价值。

主要参考文献