Polars 深度实战：当 DataFrame 遇上 Rust 与 Apache Arrow——从向量化引擎到 GPU 加速，一次把新时代数据分析讲透（2026）

一、背景：为什么 Pandas 不够用了

如果你用 Python 做过数据分析，99% 的概率你接触过 Pandas。它几乎是 Python 数据生态的代名词——从数据清洗、特征工程到金融量化分析，Pandas 的 DataFrame 无处不在。

但只要你处理过稍微大一点的数据，比如超过内存 50% 的 CSV、几千万行的日志、或者实时流式数据，你一定遇到过这些痛点：

• 单线程执行：Pandas 的操作大部分是单线程的，你的 M4 Max 有 16 个核心，但 Pandas 只用了一个
• 内存爆炸：哪怕只是读一个 2GB 的 CSV，Pandas 的峰值内存使用可能冲到 8-10GB
• 链式操作效率低：.loc[].groupby().apply() 每步都在创建中间副本
• 混乱的索引设计：df[df['a'] > 0] 和 df.loc[df['a'] > 0] 行为不一样，新手老手都踩过坑
• 应用层循环：df.apply(lambda x: custom_func(x), axis=1) 本质上还是在 Python 层面逐行循环

2026 年的数据分析场景已经完全不同了。单机数据量从 MB 涨到了 GB 甚至 TB，实时分析、流式处理、GPU 加速成了标配。我们不能再用 2008 年设计的库来处理 2026 年的数据。

这就是 Polars 诞生和爆发的根本原因。

Polars 的定位

Polars 不是一个"Pandas Drop-in Replacement"。简单地把 pd.read_csv 改成 pl.read_csv 只是最浅层的迁移。Polars 的核心理念是：把 DataFrame 当作数据库查询来优化，而不是当作 Excel 表格来操作。

它的设计哲学可以总结为四句话：

• Rust 内核，Python 皮肤——核心计算全部在 Rust 中执行，Python 只是 DSL
• 列式存储 + SIMD——天然适配现代 CPU 和 GPU 的向量化处理
• 查询优化器自动兜底——不管你写得多烂，Polars 会尽量优化执行计划
• 零拷贝、延迟执行——能不做的事坚决不做

2026 年 7 月，Polars 在 GitHub 上的 Star 数已经突破 85k，成为 Python 生态中增长最快的数据处理库。它的主要版本已经推进到 1.16.x，API 彻底稳定，GPU 支持进入 Open Beta 阶段。今天这篇文章，我们从架构原理讲到生产实战，一次把 Polars 讲透。

二、核心架构：Polars 从底层怎么工作的

要理解 Polars 为什么快，首先要搞清楚它的四层架构。

2.1 Apache Arrow：列式内存格式

Polars 的核心内存模型是 Apache Arrow，一种跨语言的列式内存格式。和 Pandas 基于 NumPy 的行式存储不同，Arrow 按列组织数据：

Pandas（行式存储）：
Row 1: [name_str, age_int, salary_float]
Row 2: [name_str, age_int, salary_float]
...

Polars / Arrow（列式存储）：
Column "name":  连续的内存块 [str, str, str, ...]
Column "age":   连续的内存块 [int, int, int, ...]
Column "salary": 连续的内存块 [float, float, float, ...]

这个差异带来了巨大的性能优势：

1. CPU 缓存友好：算平均值时只需要扫描 salary 列，不需要加载 name 列，缓存命中率极高
2. 向量化运算：连续内存天然适合 SIMD 指令，一条 CPU 指令可以同时处理 8 个 float
3. 零拷贝共享：不同进程、不同语言之间可以共享 Arrow 内存，不需要序列化
4. 谓词下推：filter(col > 100) 可以直接在列式存储上跳过多余数据块

更重要的是，Polars 不是基于 PyArrow 的封装。Polars 团队用 Rust 从头实现了自己的 Arrow 兼容层，精确控制每一个内存分配和缓存行。

// Polars 内部的 Arrow 数组（简化示意）
pub struct PrimitiveArray<T: PolarsNumericType> {
    // 连续的值缓冲区，没有 Python 对象开销
    values: Vec<T::Native>,
    // 可选的有效性位图（处理 null）
    validity: Option<Bitmap>,
    // 数据类型的元信息
    dtype: ArrowDataType,
}

这意味着什么？一个 pl.Series 中的 float64 列在内存中就是一段连续的 8 字节对齐的 Vec<f64>，没有任何 Python 对象头开销。而 Pandas 的 pd.Series 背后是 NumPy 数组——虽然底层也是连续内存，但 Pandas 的很多操作会转化为 Python 对象。

2.2 向量化查询引擎

Polars 的查询不是逐行解释执行的，而是向量化批量执行。

假设你要做这样一个操作：

df.filter(pl.col("amount") * pl.col("price") > 1000)

Polars 的执行过程是：

1. 从列中批量加载 amount、price 两个 Arrow 数组
2. 用向量化乘法算出临时数组 amount * price
3. 用向量化比较算出布尔掩码 > 1000
4. 用掩码做批量选择（gather）

每一步都在 Rust 层面通过 SIMD 指令并行处理整个列的数据块，关键代码类似：

// 简化的向量化实现
unsafe fn multiply_and_filter(a: &[f64], b: &[f64], threshold: f64) -> Bitmap {
    let mut mask = MutableBitmap::with_capacity(a.len());
    // 一次处理 32 个元素（AVX2 的 256 位寄存器可以装 4 个 f64 × 8 lanes）
    for chunk in a.chunks(32).zip(b.chunks(32)) {
        let res = simd_mul(chunk.0, chunk.1);  // SIMD 乘法
        let cmp = simd_gt(res, threshold);       // SIMD 比较
        mask.extend_from_slice(&cmp);
    }
    mask.freeze()
}

这就是为什么 Polars 在 CPU 密集计算上能比 Pandas 快 5-50 倍——Pandas 的等价操作：

# Pandas 实际执行路径
mask = (df['amount'] * df['price']) > 1000  
# ↑ 每一步都在 Python 中创建中间 Series 对象，触发引用计数和内存分配

2.3 查询优化器（LazyFrame 的精髓）

Polars 最聪明的设计之一是 Lazy API。当你调用 df.lazy() 时，Polars 不会立即执行任何操作，而是构建一个逻辑查询计划树。

df.lazy()
  .filter(pl.col("date") > "2026-01-01")
  .group_by("category")
  .agg(pl.col("revenue").sum())
  .sort("revenue", descending=True)
  .limit(10)

执行前，查询优化器会对这个计划进行多轮优化：

谓词下推（Predicate Pushdown）：把 filter 尽可能推到读取数据的阶段，甚至推到文件扫描层面。如果数据在 Parquet 文件中，Polars 可以直接利用 Parquet 的 row group 统计信息跳过不匹配的数据块。

投影下推（Projection Pushdown）：只读取后续操作需要的列。如果最终只需要 category 和 revenue 两列，Polars 就不会浪费 I/O 去读取其他列。

重新排序：如果 group_by 后接 filter，优化器会尝试交换顺序，先过滤再聚合。

公共子表达式消除：同一个表达式出现多次时只计算一次。

我们可以用 explain() 来看优化后的物理计划：

q = df.lazy().filter(pl.col("age") > 30).group_by("city").agg(pl.col("salary").mean())
print(q.explain(optimized=True))

输出类似：

--- Optimized Physical Plan ---
Aggregate [city], agg=[mean(salary)]
  Filter: (age > 30)
    ParquetScan: path/to/file.parquet, columns=[city, age, salary], 
                 predicate_pushdown: age > 30, projection_pushdown: true

注意最后一行——Polars 已经把 filter 推进了 Parquet 扫描层。这意味 80% 的数据行可能在文件扫描阶段就被跳过了，内存中从不需要加载它们。

2.4 多线程与并行调度

Polars 使用 Rust 的 Rayon 数据并行库来调度工作。默认情况下，它会使用所有可用的 CPU 核心。

这和 Python 的 GIL（全局解释器锁）形成鲜明对比——Polars 的 .group_by().agg() 不是在 Python 层面一个个 group 处理的，而是在 Rust 层面把数据切分成多个分块，用多个线程并行计算，最后合并结果。

# 这段代码在多核机器上会自动并行执行 8-16 个线程
result = df.lazy() \
    .group_by("department") \
    .agg([
        pl.col("salary").mean().alias("avg_salary"),
        pl.col("salary").max().alias("max_salary"),
        pl.col("employee_id").count().alias("headcount"),
    ]) \
    .collect()  # 触发执行

内部实现中，Polars 的 group_by 使用了 多阶段聚合（two-phase aggregation）：每个线程先对自己分到的数据块做局部哈希聚合，然后把局部哈希表合并成全局结果。

2.5 流式处理（Streaming）

当数据量超过内存时，Polars 不会直接崩溃。它的 Streaming API 可以把数据分成多个批次处理，每个批次单独计算，最后汇总结果：

# Streaming 模式：处理比内存大的数据
result = df.lazy() \
    .group_by("date") \
    .agg(pl.col("sales").sum()) \
    .collect(streaming=True)  # 流式处理

在 streaming 模式下，Polars 不是一次性加载所有数据。它会：

1. 将数据源分成多个 chunk（默认每个 50000 行）
2. 逐个 chunk 读取、过滤、聚合
3. 用外部排序或哈希表合并中间结果
4. 维持总内存使用在可控范围内

2.6 GPU 支持

2026 年，Polars 的 GPU 支持已经进入 Open Beta 阶段。通过 cuDF 兼容层或者 Polars 自有的 CUDA 内核（针对 NVIDIA GPU），在 GPU 上执行某些操作可以再快 10 倍以上：

# 启用 GPU 加速
pl.Config.set_gpu(true)

# 在 GPU 上执行，但 API 完全不变
result = df.lazy().filter(pl.col("value") > 100).collect()

目前 GPU 加速覆盖了常见的过滤、聚合、排序、join 等操作。对于内存常驻的大数据集（GPU 显存能放下），GPU 加速的性价比极高。

三、核心概念：你必须懂的 Polars 设计范式

Polars 和 Pandas 的 API 看起来很像，但设计思想完全不同。理解以下几个核心概念，是写对代码、写快代码的关键。

3.1 Expression（表达式）

在 Polars 中，几乎所有的操作都是通过 Expression 完成的。Expression 是一个惰性的、可组合的计算描述：

import polars as pl

# 这是一个 Expression，不是计算结果
expr = pl.col("salary").mean().alias("avg_salary")

Expression 的几个关键特性：

1. 可组合性

expr = (
    pl.col("price")
    .mul(pl.col("quantity"))   # price * quantity
    .floordiv(100)             # 除以 100
    .round(2)
    .alias("total_discounted")
)

2. 上下文相关

同一个 Expression 在不同的上下文中行为不同：

# select 上下文：对整列操作
df.select([
    pl.col("salary").mean().alias("avg"),
    pl.col("salary").std().alias("std"),
])

# group_by 上下文：在每个组内操作
df.group_by("department").agg([
    pl.col("salary").mean().alias("avg"),  # ❌ 组内平均值
    pl.col("salary").sum().alias("total"),
])

# over 上下文（窗口函数）：在每个窗口内操作
df.with_columns([
    pl.col("salary").mean().over("department").alias("dept_avg")
])

3. 内联 Python 函数

必要时可以用 map_elements 调用自定义 Python 函数：

df.with_columns([
    pl.col("name")
    .str.to_uppercase()
    .map_elements(lambda s: s.strip(), return_dtype=pl.Utf8)
    .alias("name_clean")
])

⚠️ 但这种方式会退化到 Python 层面逐元素执行，能不用就不用。 能用 Polars 原生表达式的就用原生表达式。

3.2 Context（上下文）

Polars 中只有 4 种上下文，理解了它们就理解了 Polars 90% 的 API：

# select：从数据中选择或派生新列
df.select([
    "name",
    pl.col("age").mean().alias("avg_age"),  # 这里 mean() 返回标量
])

# with_columns：添加新列，保持行数不变
df.with_columns([
    (pl.col("price") * pl.col("qty")).alias("total"),
    pl.col("date").dt.year().alias("year"),
])

# group_by + agg：分组聚合
df.group_by("category").agg([
    pl.col("revenue").sum(),
    pl.col("count").mean(),
    pl.col("id").n_unique().alias("unique_users"),
])

3.3 Lazy vs Eager

Polars 提供两种模式：

• Eager（即时执行）：pl.DataFrame 上的所有操作立即执行
• Lazy（惰性执行）：pl.LazyFrame 上的操作先构建查询计划，调用 .collect() 时优化并执行

# Eager 模式——和 Pandas 有点像，立即可见
df = pl.DataFrame({"x": [1, 2, 3]})
result = df.filter(pl.col("x") > 1)

# Lazy 模式——构建查询计划，延迟执行
lazy = df.lazy().filter(pl.col("x") > 1)
result = lazy.collect()  # 这里才真正执行

生产环境几乎永远应该用 Lazy API。理由很简单：

• 查询优化器可以帮你做谓词下推、投影下推
• 多个 lazy 操作可以组合后统一优化
• 可以通过 explain() 验证执行计划
• 可以启用 streaming 模式处理超内存数据

3.4 严格 Schema

Polars 是 Schema 严格的。这意味着在数据加载阶段它就确定了每一列的数据类型，并且会拒绝隐式的类型转换。

# Pandas 的惰性类型推断
pd.read_csv("data.csv")  # 类型是推断的，不保证一致

# Polars 的严格推断
pl.read_csv("data.csv", schema_overrides={
    "user_id": pl.Utf8,      # 强制字符串类型
    "amount": pl.Float64,    # 不 int 推断
})

这看起来是个小差异，但实际项目中，Polars 的严格 schema 在以下场景中价值巨大：

• 生产 pipeline 中，列类型一致性保证
• 不同批次的数据拼接时类型不冲突
• 查询优化器可以基于 schema 做更激进的优化

四、代码实战：从入门到生产

理论说够了，来写代码。

4.1 安装与基础操作

pip install polars
# 可选：安装全部可选依赖
pip install "polars[all]"

加载数据

import polars as pl

# 从 CSV 读取
df = pl.read_csv("sales_2026.csv", try_parse_dates=True)

# 从 Parquet 读取（10 倍于 CSV 的速度）
df = pl.read_parquet("sales_2026.parquet")

# 从数据库读取
df = pl.read_database_uri(
    "SELECT * FROM orders WHERE date > '2026-01-01'",
    "postgresql://user:pass@localhost:5432/db"
)

基础数据探索

# 查看前 5 行
print(df.head())

# 查看 schema
print(df.schema)
# Schema: {'order_id': Int64, 'customer': Utf8, 'amount': Float64, ...}

# 描述性统计
print(df.describe())

# 空值统计
print(df.null_count())

# 唯一值计数
print(df.unique())

# 行数 + 列数
print(df.shape)

4.2 数据清洗实战

处理缺失值

# 删除含空值的行
df_clean = df.drop_nulls()

# 按列填充空值
df_filled = df.with_columns([
    pl.col("amount").fill_null(0),
    pl.col("category").fill_null("unknown"),
    pl.col("age").fill_null(pl.col("age").median()),  # 用中位数填充
])

# 前向填充（时间序列常用）
df_ffill = df.with_columns([
    pl.col("price").forward_fill()
])

# 插值
df_interp = df.with_columns([
    pl.col("temperature").interpolate()
])

类型转换

df = df.with_columns([
    pl.col("date_str").str.strptime(pl.Date, "%Y-%m-%d").alias("date"),
    pl.col("amount_str").cast(pl.Float64).alias("amount"),
    pl.col("is_active").cast(pl.Boolean),
    pl.col("category").cast(pl.Categorical),
])

字符串处理

df = df.with_columns([
    pl.col("email").str.to_lowercase(),
    pl.col("phone").str.replace_all(r"\D", ""),  # 只保留数字
    pl.col("name").str.strip().str.to_titlecase(),
    pl.col("address").str.contains("北京").alias("is_beijing"),
    pl.col("url").str.extract(r"/product/(\d+)", 1).alias("product_id"),
])

日期时间处理

df = df.with_columns([
    pl.col("date").dt.year().alias("year"),
    pl.col("date").dt.month().alias("month"),
    pl.col("date").dt.quarter().alias("quarter"),
    pl.col("date").dt.weekday().alias("weekday"),
    pl.col("timestamp").diff().alias("time_diff"),
])

4.3 聚合分析实战

多维度分组聚合

result = (
    df.lazy()
    .group_by(["year", "category", "region"])
    .agg([
        pl.col("revenue").sum().alias("total_revenue"),
        pl.col("revenue").mean().alias("avg_revenue"),
        pl.col("revenue").std().alias("std_revenue"),
        pl.col("cost").sum().alias("total_cost"),
        (pl.col("revenue").sum() - pl.col("cost").sum()).alias("profit"),
        pl.col("order_id").count().alias("order_count"),
        pl.col("customer_id").n_unique().alias("unique_customers"),
        pl.col("quantity").quantile(0.9).alias("p90_quantity"),
    ])
    .sort("total_revenue", descending=True)
    .collect()
)

滚动聚合（时间窗口）

result = (
    df.lazy()
    .sort("date")
    .with_columns([
        # 7 天滚动平均
        pl.col("revenue").rolling_mean(window_size=7).alias("revenue_ma7"),
        # 30 天滚动求和
        pl.col("orders").rolling_sum(window_size=30).alias("orders_sum30"),
        # 按组的滚动聚合
        pl.col("amount")
        .rolling_mean(window_size=3)
        .over("product_id")
        .alias("product_ma3"),
    ])
    .collect()
)

窗口函数

result = (
    df.lazy()
    .with_columns([
        # 各部门内的薪资排名
        pl.col("salary").rank("dense").over("department").alias("dept_salary_rank"),
        # 各部门薪资的最大差距
        (pl.col("salary") - pl.col("salary").min().over("department")).alias("salary_premium"),
        # 累计求和
        pl.col("revenue").cum_sum().over("category").alias("category_cumulative"),
        # 各部门薪资占比
        (pl.col("salary") / pl.col("salary").sum().over("department")).alias("salary_pct"),
    ])
    .collect()
)

4.4 Join 操作

Polars 的 join 同样是用向量化实现的，比 Pandas 的 join 快 10-50 倍：

# 左连接
result = orders.lazy().join(
    customers.lazy(),
    left_on="customer_id",
    right_on="id",
    how="left"
).collect()

# 反连接（在左边但不在右边）
result = orders.lazy().join(
    cancelled.lazy(),
    left_on="order_id",
    right_on="order_id",
    how="anti"
).collect()

# 半连接（只要匹配的行）
result = orders.lazy().join(
    vip_customers.lazy(),
    left_on="customer_id",
    right_on="id",
    how="semi"
).collect()

# 交叉连接（笛卡尔积）
result = products.lazy().join(
    stores.lazy(),
    how="cross"
).collect()

Polars 对于 join 的实现使用了 分治哈希连接（divide-and-conquer hash join）。当两个表都很大时，它会先用左表的哈希分布切分数据，然后多线程并行构建哈希表，最后合并结果。

4.5 复杂数据分析案例

来一个真实的生产级案例：电商用户行为漏斗分析。

# 加载点击流数据
clicks = pl.scan_parquet("user_clicks_2026*.parquet")
orders = pl.scan_parquet("orders_2026*.parquet")
users = pl.scan_parquet("users_2026*.parquet")

# 构建用户行为漏斗
funnel = (
    clicks
    .filter(pl.col("event_type").is_in(["page_view", "add_to_cart", "checkout", "purchase"]))
    .group_by(["user_id", "event_type"])
    .agg(pl.col("event_id").count().alias("event_count"))
    .join(users, on="user_id", how="left")
    .filter(pl.col("is_active") == True)
    .group_by("event_type")
    .agg([
        pl.col("user_id").n_unique().alias("unique_users"),
        pl.col("event_count").sum().alias("total_events"),
        pl.col("event_count").mean().alias("avg_events_per_user"),
    ])
    .with_columns([
        # 计算每一步的转化率
        (pl.col("unique_users") / pl.col("unique_users").shift(1)).alias("step_conversion"),
        # 计算相对于首步的总转化率
        (pl.col("unique_users") / pl.col("unique_users").first()).alias("overall_conversion"),
    ])
    .sort(pl.col("step_conversion").rank("ordinal"))
    .collect()
)

这段代码在 Polars 中会：

1. 利用 Parquet 的谓词下推只读需要的列
2. 多线程并行执行两个 group_by
3. 窗口函数（shift、first）用向量化执行
4. 整个过程中间结果不会物化到完整内存，streaming 模式下几乎零拷贝

在 Pandas 里实现等价的逻辑可能需要 3-5 个中间 DataFrame，内存翻倍。

4.6 SQL 接口

如果你熟悉 SQL，Polars 也提供了 SQL 查询接口：

# 注册 DataFrame 为 SQL 表
pl.SQLContext().register("orders", orders_df)
pl.SQLContext().register("customers", customers_df)

# 直接用 SQL 查询
result = pl.SQLContext().execute("""
    SELECT 
        c.region,
        EXTRACT(YEAR FROM o.order_date) AS year,
        SUM(o.amount) AS total_revenue,
        COUNT(DISTINCT o.customer_id) AS unique_customers
    FROM orders o
    JOIN customers c ON o.customer_id = c.id
    WHERE o.amount > 0
    GROUP BY c.region, year
    HAVING total_revenue > 10000
    ORDER BY total_revenue DESC
""")

这很实用——当你的同事更熟悉 SQL 时，可以先用 DataFrame API 做数据清洗，再用 SQL 做分析，最终结果仍然是 Polars 的 DataFrame。

五、性能优化：把 Polars 的性能榨干

5.1 优化法则一：永远用 Lazy API

这是最重要的一条。Lazy API 让优化器替你工作。

# ❌ 坏习惯：每次操作都 collect
df = pl.read_csv("data.csv")
df = df.filter(pl.col("x") > 0)
df = df.group_by("cat").agg(pl.col("y").sum())
df = df.collect()  # 这个多余的 collect 毫无意义

# ✅ 正确做法：一气呵成
result = (
    pl.scan_csv("data.csv")   # scan 而非 read，这本身就是 lazy 的
    .filter(pl.col("x") > 0)
    .group_by("cat")
    .agg(pl.col("y").sum())
    .collect()
)

pl.scan_csv() vs pl.read_csv() 的区别非常关键——scan 是惰性的，它会延迟读取文件直到 .collect() 被调用，而且可以在这个阶段做谓词下推。

5.2 优化法则二：充分利用 Parquet

Parquet 是 Polars 的最佳搭档。它的列式存储 + 压缩 + row group 统计信息，让 Polars 可以跳过大量无关数据。

# 从 CSV 转 Parquet
pl.read_csv("huge.csv").write_parquet("huge.parquet")

# 然后使用 partitioned Parquet（按日期分区）
(
    pl.scan_parquet("data/year=2026/month=*/")
    .filter(pl.col("region") == "华东")
    .collect()
)

Parquet 相比 CSV 的优势：

• 存储空间减少 70-90%（列式压缩）
• 读取速度提升 10-50 倍（只读需要的列）
• 元数据统计可以跳过不匹配的 row group

5.3 优化法则三：选择正确的数据类型

# ❌ 低效类型
pl.Series("id", ["1", "2", "3"], dtype=pl.Utf8)     # 字符串，12+ 字节/元素

# ✅ 高效类型
pl.Series("id", [1, 2, 3], dtype=pl.Int32)           # 整数，4 字节/元素

# 如果值很少，用 Categorical
pl.Series("status", ["active", "inactive", "active"], dtype=pl.Categorical)
# Categorical 内部用整数表示，大幅节省内存和加速 group_by

类型选择对性能的影响非常具体：

5.4 优化法则四：控制 partition 数量

Polars 默认的并行度是 CPU 核心数。对于某些操作，你可能需要手动控制：

# 限制并行线程数（避免和其他服务争抢 CPU）
pl.Config.set_thread_pool_size(4)

# 控制 streaming 的 batch size
result = df.lazy().group_by("key").agg(pl.col("value").sum()).collect(
    streaming=True,
    new_streaming=True,  # 新版 streaming 引擎
)

5.5 优化法则五：用 Expression Plugin 扩展性能

Polars 支持用 Rust 编写自定义表达式插件，获得原生级别的性能。你不需要编译 Polars 本身，只需要写一个独立的 Rust crate：

// 一个自定义的 Polars 表达式插件
use polars::prelude::*;
use polars_plugin::*;

#[polars_expr(output_type=Float64)]
fn custom_scaling(inputs: &[Series]) -> PolarsResult<Series> {
    let s = &inputs[0];
    let min = s.min_as_series().get(0).try_extract::<f64>()?;
    let max = s.max_as_series().get(0).try_extract::<f64>()?;
    
    // 向量化 Min-Max 缩放
    let result = s.f64()?
        .apply_values(|v| (v - min) / (max - min));
    
    Ok(result.into_series())
}

然后在 Python 中调用：

df.with_columns([
    pl.col("score").map_batches(
        lambda s: custom_scaling(s),
        return_dtype=pl.Float64,
        is_elementwise=True,
    )
])

这适用于标准库不能覆盖的自定义高性能操作。

六、生产实战：Polars 在大规模系统中的落地

6.1 ETL 管道中的 Polars

import polars as pl
from pathlib import Path
import logging

def etl_pipeline(input_path: Path, output_path: Path, db_url: str):
    """生产级 ETL 管道"""
    
    # 1. 并行扫描所有输入文件
    raw = pl.scan_parquet(str(input_path / "*.parquet"))
    
    # 2. 数据清洗和转换（全 lazy）
    cleaned = (
        raw
        .filter(
            pl.col("status").is_not_null() &
            pl.col("amount") > 0
        )
        .with_columns([
            pl.col("date").str.strptime(pl.Date, "%Y-%m-%d"),
            pl.col("amount").cast(pl.Float64),
            pl.col("category").fill_null("unknown"),
            # 特征工程
            pl.col("amount").log1p().alias("amount_log"),
            pl.col("amount").clip(0, 10_000).alias("amount_clipped"),
        ])
        .with_columns([
            # 派生时间维度
            pl.col("date").dt.year().alias("year"),
            pl.col("date").dt.month().alias("month"),
            pl.col("date").dt.quarter().alias("quarter"),
            pl.col("date").dt.weekday().alias("weekday"),
        ])
    )
    
    # 3. 写入清洗后的 Parquet
    cleaned.sink_parquet(str(output_path / "clean"))
    
    # 4. 同时写入数据库（兼容模式）
    cleaned_df = cleaned.collect(streaming=True)
    cleaned_df.write_database(
        table_name="sales_clean",
        connection=db_url,
        if_table_exists="append",
    )
    
    logging.info(f"ETL pipeline completed: {cleaned_df.shape[0]} rows")
    return cleaned_df

注意这里使用了 .sink_parquet() —— Polars 在流式模式下的写入操作，不需要先 collect 再写。

6.2 超大文件分块处理

当文件大到连 Polars 也无法一次性处理时：

CHUNK_SIZE = 500_000

# 分块读 + 增量写
for batch in pl.read_csv_batched("extremely_large.csv", chunk_size=CHUNK_SIZE):
    batch_parsed = batch.with_columns([
        pl.col("date").str.strptime(pl.Date, "%Y-%m-%d"),
    ])
    # 追加到已存在的 Parquet 文件
    batch_parsed.write_parquet("output.parquet", append=True)

但说真的，对于大部分场景，直接用 .scan_csv() + .collect(streaming=True) 就够了，不用手写分块逻辑。

6.3 与机器学习框架集成

# Polars → NumPy → scikit-learn
from sklearn.ensemble import GradientBoostingRegressor

features = df.select([
    pl.col("feature_1", "feature_2", "feature_3"),
    pl.col("target"),
])

X = features.select(pl.all().exclude("target")).to_numpy()
y = features["target"].to_numpy()

model = GradientBoostingRegressor()
model.fit(X, y)

# Polars → PyTorch DataLoader
import torch
from torch.utils.data import DataLoader, TensorDataset

tensor_data = torch.from_numpy(
    df.select(pl.col("feature_1", "feature_2")).to_numpy()
)
tensor_target = torch.from_numpy(
    df["target"].to_numpy()
)

dataset = TensorDataset(tensor_data, tensor_target)
loader = DataLoader(dataset, batch_size=256, shuffle=True)

for batch_x, batch_y in loader:
    # training loop
    pass

6.4 监控与可观测性

# 查询计划可视化
plan = df.lazy().filter(pl.col("x") > 0).group_by("y").agg(pl.col("z").sum())

# 查看未优化前的逻辑计划
print("=== LOGICAL PLAN ===")
print(plan.explain(optimized=False))

# 查看优化后的物理计划
print("=== OPTIMIZED PHYSICAL PLAN ===")
print(plan.explain(optimized=True))

# 性能分析
import time
start = time.time()
result = plan.collect()
elapsed = time.time() - start
print(f"Query executed in {elapsed:.3f}s, result shape: {result.shape}")

七、Polars vs 竞品：如何选型

7.1 Pandas vs Polars

什么时候还用 Pandas？

• 数据量 < 100MB，性能不是瓶颈
• 重度依赖 Pandas 独有功能（如 panel data、MultiIndex 复杂操作）
• 团队全员只熟悉 Pandas，迁移成本太高
• 项目即将下线，不值得重写

什么时候切换到 Polars？

• 数据量 >= 1GB 时，性能差异就能感受到
• 新项目，没有历史包袱
• 生产 Pipeline，需要稳定、可预测的性能
• 需要处理 Parquet、流式数据
• 团队接受新工具

7.2 DuckDB vs Polars

DuckDB 也是一个崛起的分析引擎，但和 Polars 的定位不同：

• DuckDB：嵌入式 OLAP 数据库，用 SQL 查询，适合 SQL 重度用户
• Polars：程序化 DataFrame 库，以 Python DataFrame API 为主，也支持 SQL

它们不是竞争关系，而是互补关系。实际上，Polars 原生支持 DuckDB 作为后端：

# 用 DuckDB 引擎查询 Polars 的 DataFrame
result = pl.from_duckdb("""
    SELECT category, SUM(amount)
    FROM df
    GROUP BY category
""")

7.3 Spark vs Polars

PySpark 是分布式解决方案，Polars 是单机优化解决方案。

经验法则：数据量在 100GB 以下，优先用 Polars。超过这个阈值再考虑 Spark。

八、避坑指南：Polars 常见陷阱

8.1 null 处理

Polars 的 null 和 Python 的 None 不完全等价：

# ❌ 常见错误
df.filter(pl.col("value") == None)  # 不会工作！

# ✅ 正确做法
df.filter(pl.col("value").is_null())
df.filter(pl.col("value").is_not_null())

8.2 隐式广播

# Polars 不会自动广播标量
# ❌ 错误
df.with_columns(pl.col("value") - df["value"].mean())

# ✅ 正确
df.with_columns(
    (pl.col("value") - pl.col("value").mean()).alias("centered")
)

8.3 链式操作的复制

# 虽然看起来像在修改 df，但实际上每次都创建新 DataFrame
df = df.filter(pl.col("x") > 0)
df = df.with_columns(pl.lit(1).alias("y"))
df = df.group_by("y").agg(pl.col("x").sum())

# 更好的方式：Lazy API + 一次性 collect
result = (
    df.lazy()
    .filter(pl.col("x") > 0)
    .with_columns(pl.lit(1).alias("y"))
    .group_by("y")
    .agg(pl.col("x").sum())
    .collect()
)

8.4 不要频繁在 Python 和 Rust 间切换

# ❌ 极慢：每次 map_elements 都在 Python 和 Rust 间切换
df = df.with_columns(
    pl.col("value").map_elements(
        lambda x: x * 2 + 1, return_dtype=pl.Float64
    )
)

# ✅ 极快：全部在 Rust 层面执行
df = df.with_columns(
    (pl.col("value") * 2 + 1).alias("value_doubled")
)

在大量数据上使用 map_elements 时，性能差距可能达到 1000 倍。能用 Polars Expression 的就不要用 Python 函数。

九、总结与展望

Polars 不是另一个"Pandas 替代品"，它是一个重新设计的、面向现代硬件的数据分析基础设施。

它的核心优势在于：

1. 架构碾压：Rust 内核 + Apache Arrow + SIMD + 查询优化器，这四层叠加使它在单机上就能处理传统上需要 Spark 集群的任务
2. API 优雅：Expression 系统比 Pandas 的对象链式操作更清晰、更可组合、更容易调试
3. 生态兼容：与 DuckDB、PyArrow、NumPy、PyTorch 等深度互操作，不是孤岛
4. 生产就绪：流式处理、GPU 加速、表达式插件、成熟的写入端（sink）API

Polars 2026 年的五个关键趋势

趋势一：GPU 加速走向 GA。Polars 的 GPU 支持在 2026 年 Q3 达到 GA 状态，对 NVIDIA GPU 的加速可以达到 10-50 倍。对于显存能放下的数据，这几乎等价于免费的算力提升。

趋势二：Polars Cloud 商业化。Polars Cloud 提供了无服务器的 DataFrame 计算服务，支持在云上横向扩展，解决了"单机好但集群不行"的最后一个短板。

趋势三：AI/ML 原生集成。Polars 正在成为 AI 数据管道的首选中间层——从数据加载、清洗、特征工程到模型训练输入的完整通路。

趋势四：查询优化器持续进化。Polars 团队正在引入基于成本的优化器（CBO），可以根据数据的统计信息选择最优的执行策略。

趋势五：Polars 成为数据分析的 Lingua Franca。越来越多的工具（如 Evidence、Quarto、Hex）开始原生支持 Polars 作为数据后端。

给读者的建议

如果你现在还在用 Pandas 处理几百 MB 以上的数据，是时候认真考虑迁移了。不需要一次性全量迁移，可以从以下路径开始：

1. 新写的分析脚本直接用 Polars
2. 把现有的 ETL Pipeline 中的 Pandas 片段逐个替换
3. 把 CSV 存储逐步转为 Parquet
4. 在关键性能路径上启用 Lazy API + Streaming

2026 年的数据分析已经不是"能不能跑"的问题，而是"能不能在有限资源下跑得又快又好"。Polars 给出了一个优雅的答案——用现代工程重新思考一个经典问题。就像当年 NumPy 替代了纯 Python 的数值计算，Polars 正在替代 Pandas 成为数据处理的新一代标准。

去试试吧，你可能会惊讶于你的数据到底能处理多快。

# 最后的彩蛋：一行代码感受 Polars 的速度
pl.DataFrame({"x": range(10_000_000)}).with_columns(
    (pl.col("x") * 2 + pl.col("x").shift(1)).alias("computed")
).describe()

在 Pandas 里跑同样的操作，先去泡杯咖啡。在 Polars 里，你还没走到饮水机，结果已经出来了。