Polars：比Pandas更快的数据处理引擎

可视化演示

对比Polars与Pandas的性能基准、展示惰性执行查询优化流程、以及多线程并行分片计算机制，直观呈现Polars的速度优势。

Polars核心架构与Rust内存安全执行模型

Polars 是一个用 Rust 语言从零编写的高性能 DataFrame 库。Rust 的所有权系统和借用检查器在编译期就消除了数据竞争和悬空指针等问题，使得 Polars 的底层执行引擎天然具备内存安全保证。与 C/C++ 不同，Rust 无需运行时垃圾回收，也无需全局解释器锁（GIL），这意味着 Polars 可以真正利用多核 CPU 进行并行计算，而不会受到 Python 运行时单线程瓶颈的制约。在大数据量场景下，这种架构优势直接转化为数倍乃至数十倍的性能提升。

Polars 的整体架构分为三个核心层次：表达式层（Expression Layer）、查询计划层（Query Plan Layer）和执行引擎层（Execution Engine Layer）。表达式层负责将用户的链式调用转化为结构化的表达式树；查询计划层将表达式树组装为逻辑计划并进行全局优化，包括谓词下推、投影裁剪和操作融合；执行引擎层则将优化后的物理计划分配到多线程并行执行。三层解耦的设计使得每一层都可以独立演进，例如优化器可以在不影响用户 API 的前提下持续增强优化策略。

Python 的 Pandas 库由于底层基于 NumPy 和 Python 运行时，在执行过程中会产生大量临时对象和中间结果，导致内存碎片化严重。当处理数十 GB 的数据时，内存碎片化会显著增加内存分配器的压力，进而降低整体吞吐量。Polars 通过 Rust 的内存分配策略和 Arrow 列式格式，从根本上避免了这一问题，使得内存使用更加紧凑和可预测。

import polars as pl df = pl.DataFrame({    "user_id": [1, 2, 3, 4],    "score": [85.0, 92.5, 78.3, 95.1]})# Polars 底层通过 Rust 执行，无 GIL 限制result = df.filter(pl.col("score") > 80).select(pl.col("user_id"))

Apache Arrow列式内存格式与零拷贝集成

Apache Arrow 定义了一套跨语言的列式内存标准。所谓列式存储，是指同一列的数据在内存中连续排列，这与传统行式存储（同一行的数据连续排列）形成鲜明对比。列式存储的核心优势在于：当执行过滤、聚合等只需要访问少数几列的操作时，CPU 缓存可以高效地加载所需数据，而无需跳过大量无关列的数据。在实际基准测试中，列式格式在聚合场景下的缓存命中率可以达到行式格式的 5 到 10 倍。

Arrow 作为跨进程和跨语言的内存标准，使得 Polars 可以与 PyArrow、DuckDB、Spark 等 Arrow 生态工具之间实现零拷贝数据传递。零拷贝意味着数据在不同工具间流转时无需序列化和反序列化，内存中的 Arrow 缓冲区可以直接被另一个工具读取。这在构建多工具协作的数据管道时尤为重要，因为数据序列化往往是最大的性能瓶颈之一。

Polars 的 DataFrame 在内存中直接以 Arrow 格式存储，这意味着从磁盘读取 Parquet 文件（Parquet 底层使用 Arrow 列式编码）时，数据可以几乎零开销地加载到内存中。同样，当需要将处理结果写回 Parquet 文件或传递给其他 Arrow 兼容工具时，也无需格式转换。这种端到端的列式内存管理是 Polars 高性能的关键基础设施。

import pyarrow as paimport polars as pl # Arrow Table 与 Polars DataFrame 零拷贝互转arrow_table = pa.table({"x": [1, 2, 3], "y": [4.0, 5.0, 6.0]})df = pl.from_arrow(arrow_table)  # 零拷贝back_to_arrow = df.to_arrow()    # 零拷贝

惰性执行引擎与查询优化器原理

惰性执行（Lazy Evaluation）是 Polars 区别于 Pandas 的最核心特性。在 Pandas 中，每一步操作都会立即执行并产生中间结果，例如 df.filter(……).groupby(……).agg(……) 会依次执行过滤、分组、聚合三步，每步都生成新的 DataFrame 对象。而 Polars 的惰性 API 会将所有操作收集起来构建一棵逻辑查询计划树，直到用户显式调用 collect() 方法时才统一交给查询优化器处理。

查询优化器在拿到完整的逻辑计划后，会执行三类关键优化。第一是谓词下推（Predicate Pushdown），将过滤条件尽可能推到数据扫描阶段，从而在数据进入计算管道之前就减少数据量。第二是投影裁剪（Projection Pushdown），只读取查询实际需要的列，跳过无关列以减少 I/O 和内存占用。第三是操作融合（Operation Fusion），将连续的过滤或映射操作合并为单次遍历，避免多次扫描数据和中间结果的物化开销。

可以用一个类比来理解惰性执行的价值：就像快递公司的智能路线规划系统，你把所有要送的包裹和目的地一次性输入系统，系统会全局规划最优配送路线，合并顺路订单，而不是每来一个包裹就立刻派一辆车出去跑一趟。在复杂数据管道中，惰性执行带来的优化效果可以达到数倍甚至数十倍的性能差异。

import polars as pl # 惰性 API：构建查询计划但不立即执行lazy_df = (    pl.scan_parquet("data.parquet")    .filter(pl.col("date") >= "2024-01-01")    .groupby("category")    .agg(pl.col("revenue").sum()))# 调用 collect() 时才触发优化器并执行result = lazy_df.collect()

多线程并行计算与流式处理机制

Polars 的并行计算建立在 Rust 的无锁并发基础之上。当执行一个聚合操作时，Polars 会将数据按行分片（partition）成多个子集，每个子集分配给一个独立的线程处理。由于 Rust 的所有权系统在编译期就保证了数据访问的互斥性，Polars 无需使用运行时锁来协调线程，从而避免了锁竞争带来的性能损耗。在 32 核服务器上，聚合操作的加速比通常可以达到 20 到 28 倍。

对于超出可用内存的超大数据集，Polars 提供了流式处理引擎（Streaming Engine）。流式引擎不会一次性将整个数据集加载到内存，而是按固定大小的批次（默认约 256MB）读取和处理数据。每个批次在内存中完成计算后，中间结果被写入磁盘或发送给下游算子，然后释放内存供下一个批次使用。这种模式就像自助餐厅的流水线洗碗机，不需要把所有脏碗碟一次性堆进机器，而是按批连续送入清洗，机器始终满负荷运转但占用的空间始终只有一批碗碟的大小。

流式处理的关键技术包括分块聚合（Chunked Aggregation）和外部排序（External Sort）。分块聚合将全局聚合分解为每个批次内的局部聚合，最后合并局部结果得到全局结果。外部排序则先将数据分块排序写入临时文件，再通过多路归并完成全局排序。这些技术使得 Polars 可以在 16GB 内存的机器上稳定处理数百 GB 的数据集。

import polars as pl # 流式执行：适合超大数据集result = (    pl.scan_parquet("huge_data.parquet")    .groupby("user_id")    .agg(pl.col("amount").sum())    .collect(streaming=True)  # 启用流式引擎)

表达式API设计理念与Pandas迁移策略

Polars 的表达式 API 采用声明式编程范式，用户通过链式调用描述想要什么结果，而不是如何一步步得到结果。例如 pl.col("price").mul(pl.col(“quantity“)).sum() 表达的是”计算价格乘以数量后的总和”这一意图，而不需要指定循环或索引方式。这种声明式风格使得查询意图清晰可读，同时也便于优化器分析和优化整个表达式树。

从 Pandas 迁移到 Polars 时，需要理解几个关键语义差异。首先，Polars 没有 Pandas 的 Index 概念，数据定位通过列名和布尔表达式完成，而非位置索引。其次，Pandas 的 groupby 返回的是 Series 或 DataFrame，而 Polars 的 groupby 必须配合 agg 表达式使用，语法上更加严格但语义更加明确。第三，Pandas 的链式操作是即时执行的，每步返回新对象，而 Polars 的惰性链式操作构建的是查询计划，只在 collect 时执行。

推荐的迁移策略是渐进式迁移而非一次性重写。首先在新项目中直接使用 Polars 的惰性 API，对于已有 Pandas 代码，可以通过 Arrow 格式在 Pandas 和 Polars 之间零拷贝转换数据，逐步将性能瓶颈模块替换为 Polars 实现。在迁移过程中，重点理解表达式链式调用的思维方式，而非逐行翻译 Pandas 代码。

import polars as pl # Pandas 风格（不推荐）: df[df['score'] > 80]['name']# Polars 表达式风格（推荐）:result = (    pl.DataFrame({"name": ["A", "B", "C"], "score": [85, 70, 92]})    .filter(pl.col("score") > 80)    .select(pl.col("name")))

AI数据管道中的Polars性能优化实践

在 AI 数据管道中，选择合适的 API 模式是性能优化的第一步。惰性 API（LazyFrame）能够启用查询优化器的全部优化策略，包括谓词下推、投影裁剪和操作融合，在复杂多步管道中性能优势显著。即时 API（Eager API）虽然使用体验更接近 Pandas，但每步操作立即执行，无法享受全局优化。生产环境中应优先使用惰性 API，仅在需要交互式调试或简单操作时使用即时 API。

分片数量的设置直接影响并行计算的效率。Polars 默认根据 CPU 核心数自动设置分片数量，但在某些场景下手动调整可以获得更好的性能。分片数过少会导致部分核心空闲，分片数过多则会增加线程调度和数据合并的开销。经验法则是将分片数设置为物理核心数的 1 到 2 倍，并通过基准测试验证不同配置下的性能差异。

延迟加载（Lazy Loading）是处理超大数据文件的关键策略。使用 scan_parquet 或 scan_csv 而非 read_parquet 或 read_csv，可以让优化器在查询计划阶段就确定需要读取的数据范围，从而只加载必要的数据。此外，减少 Python 与 Rust 边界的数据转换也是重要的优化方向。在热路径中频繁调用 apply 或 map_rows 等需要逐行调用 Python 函数的操作，会引入显著的跨边界开销，应尽量用 Polars 原生表达式替代。

import polars as pl # 优化实践：惰性加载 + 谓词下推 + 投影裁剪result = (    pl.scan_parquet("train_data.parquet")    .filter(pl.col("label").is_not_null())    .select(["feature_1", "feature_2", "label"])    .groupby("label")    .agg([        pl.col("feature_1").mean(),        pl.col("feature_2").std()    ])    .collect(streaming=True))

AI 应用场景

Python 演示

import numpy as np # === 模拟 Polars 核心机制: 惰性求值、表达式API、并行执行、流式处理 === # 1. 表达式节点定义class Expr:    def __init__(self, op, col=None, value=None, left=None, right=None):        self.op = op        self.col = col        self.value = value        self.left = left        self.right = right def col(name):    return Expr("col", col=name) def lit(value):    return Expr("lit", value=value) # 2. 惰性查询计划class LazyPlan:    def __init__(self, source):        self.steps = []        self.source = source    def filter(self, expr):        self.steps.append(("filter", expr))        return self    def select(self, exprs):        self.steps.append(("select", exprs))        return self    def groupby_agg(self, key_col, agg_expr):        self.steps.append(("groupby_agg", key_col, agg_expr))        return self # 3. 执行引擎 - 解析表达式并求值def eval_expr(expr, data):    if expr.op == "col":        return data[expr.col]    elif expr.op == "lit":        n = len(next(iter(data.values())))        return np.full(n, expr.value)    elif expr.op == "gt":        return eval_expr(expr.left, data) > expr.value    elif expr.op == "sum":        return np.array([np.sum(eval_expr(expr.left, data))])    elif expr.op == "mean":        return np.array([np.mean(eval_expr(expr.left, data))]) # 4. 优化器 - 谓词下推: 将filter提前到select之前def optimize(plan):    filters = [s for s in plan.steps if s[0] == "filter"]    others = [s for s in plan.steps if s[0] != "filter"]    return filters + others # 5. 并行执行模拟 - 数据分片后多线程处理def parallel_apply(data, func, n_splits=4):    n = len(next(iter(data.values())))    indices = np.array_split(np.arange(n), n_splits)    results = []    for idx in indices:        chunk = {k: v[idx] for k, v in data.items()}        results.append(func(chunk))    return results # 6. 流式处理模拟 - 按批次读取和处理def streaming_collect(plan, data_source, batch_size=3):    optimized = optimize(plan)    all_results = []    n = len(data_source["id"])    for start in range(0, n, batch_size):        end = min(start + batch_size, n)        batch = {k: v[start:end] for k, v in data_source.items()}        for step in optimized:            if step[0] == "filter":                mask = eval_expr(step[1], batch)                batch = {k: v[mask] for k, v in batch.items()}            elif step[0] == "select":                batch = {e.col: eval_expr(e, batch) for e in step[1]}        all_results.append(batch)    return all_results # === 运行演示 ===np.random.seed(42)N = 12source = {    "id": np.arange(N),    "category": np.array(["A","B","A","B","A","B","A","B","A","B","A","B"]),    "value": np.random.randn(N) * 10 + 50} # 构建惰性计划plan = LazyPlan(source).filter(col("value").__class__("gt", left=col("value"), value=45)).select([col("category"), col("value")]) # 流式执行batches = streaming_collect(plan, source, batch_size=4)print("Streaming batches:", len(batches))for i, b in enumerate(batches):    print(f"  Batch {i}: {len(b.get('id', b.get('category', [])))} rows") # 并行聚合chunks = parallel_apply(source, lambda d: np.sum(d["value"]))total = sum(chunks)print(f"Parallel sum: {total:.2f}")

1. class Expr:定义表达式节点类，每个节点表示一个操作（如取列、比较、聚合），多个节点组成表达式树，这是 Polars 表达式 API 的基础数据结构。
2. def __init__(self, op, col=None, value=None, left=None, right=None):表达式节点通过 op 字段标识操作类型，col 指定列名，value 存储字面量，left/right 指向子节点，形成树状结构。
3. class LazyPlan:惰性查询计划类，不立即执行操作，而是将 filter、select、groupby_agg 等步骤记录到 steps 列表中，延迟到 collect 时统一优化执行。
4. def filter(self, expr):将过滤条件追加到查询计划中，返回 self 支持链式调用，此时不执行任何数据操作。
5. def eval_expr(expr, data):表达式求值函数，递归遍历表达式树，根据 op 类型执行取列、字面量广播、比较或聚合等操作，返回 numpy 数组。
6. def optimize(plan):简化版查询优化器，将 filter 步骤重新排列到 select 之前，模拟谓词下推优化，减少后续操作需要处理的数据量。
7. def parallel_apply(data, func, n_splits=4):模拟 Polars 的并行执行：将数据按行分成 n_splits 个分片，每个分片独立执行计算函数，最后合并结果，模拟多线程并行聚合。
8. indices = np.array_split(np.arange(n), n_splits)将数据行索引均匀分割为多个分片，每个分片对应一个线程的处理范围，这是 Polars 并行计算的数据分片策略。
9. def streaming_collect(plan, data_source, batch_size=3):流式处理函数，按 batch_size 分批读取数据，每批独立执行优化后的查询计划，避免一次性加载全部数据到内存。
10. batch = {k: v[start:end] for k, v in data_source.items()}从数据源中按起止索引截取当前批次的数据，模拟 Polars 流式引擎的分批读取行为。
11. mask = eval_expr(step[1], batch)在当前批次上执行过滤表达式，生成布尔掩码，实现谓词下推后的数据过滤。
12. plan = LazyPlan(source).filter(col("value").__class__("gt", left=col("value"), value=45)).select([col("category"), col("value")])构建惰性查询计划：先过滤 value 大于 45 的行，再选择 category 和 value 列，所有操作延迟执行。
13. batches = streaming_collect(plan, source, batch_size=4)以流式方式执行查询计划，每批处理 4 行数据，展示 Polars 如何在有限内存下处理超大数据集。

常见问答

常见误区

参考来源

• docPolars Official DocumentationPolars 官方文档，涵盖核心架构、惰性执行引擎、表达式 API、流式处理等核心概念和 API 参考。
• docApache Arrow DocumentationApache Arrow 官方文档，详细说明列式内存格式规范、零拷贝数据传递机制和跨语言生态集成。
• webpagePolars GitHub RepositoryPolars 开源代码仓库，包含 Rust 底层实现源码、性能基准测试和从 Pandas 迁移的指南。
• docPolars Lazy API OptimizationsPolars 惰性 API 优化策略文档，详细介绍谓词下推、投影裁剪和操作融合的原理与配置。