Polars 深度解析：用 Rust 重写数据分析规则，比 Pandas 快 10-100 倍的秘密

当 Python 数据分析社区还在为 Pandas 的内存溢出和单线程瓶颈头疼时，一个用 Rust 编写的新秀已经悄悄拿下了 80k+ GitHub Star。Polars——基于 Apache Arrow 的列式 DataFrame 库，正在重新定义数据分析的性能天花板。本文深度解析 Polars 的架构设计、惰性计算引擎、多线程并行策略，以及从 Pandas 迁移的完整实战指南。

一、为什么需要 Polars？

1.1 Pandas 的三大痛点

Pandas 是 Python 数据分析的基石，但随着数据规模从 MB 时代进入 GB 甚至 TB 时代，它的架构瓶颈越来越明显：

痛点 1：单线程执行

import pandas as pd
import time

# 1000 万行数据过滤
df = pd.DataFrame({
    'id': range(10_000_000),
    'value': np.random.randn(10_000_000),
    'category': np.random.choice(['A', 'B', 'C', 'D'], 10_000_000)
})

start = time.time()
result = df.groupby('category').agg({'value': ['mean', 'sum', 'std']})
print(f"Pandas 耗时: {time.time() - start:.2f}s")
# 输出: Pandas 耗时: 1.83s  ← 只用了一个 CPU 核心

痛点 2：内存浪费

Pandas 基于行式存储（NumPy ndarray），每次操作都可能创建中间副本：

# Pandas 的内存陷阱
df = pd.read_csv('large_dataset.csv')  # 1GB 文件
print(f"加载后内存: {df.memory_usage(deep=True).sum() / 1e9:.2f} GB")
# 输出: 加载后内存: 2.8 GB  ← 近 3 倍膨胀！

# 链式操作每次都创建中间 DataFrame
result = (df
    .query('age > 25')           # 副本 1
    .groupby('city')             # 副本 2
    .agg({'salary': 'mean'})     # 副本 3
)
# 峰值内存可能达到 5-8 GB

痛点 3：类型推断不一致

# Pandas 的类型陷阱
df = pd.DataFrame({'id': [1, 2, None, 4]})
print(df['id'].dtype)  # float64 ← 因为 None，整数列变成了浮点数！

# 混合类型列更惨
df = pd.DataFrame({'col': ['1', '2', 'three', '4']})
print(df['col'].dtype)  # object ← 退化为 Python 对象，性能灾难

1.2 Polars 的答案

二、核心架构：为什么 Polars 这么快？

2.1 Apache Arrow 列式内存模型

Polars 的内存模型基于 Apache Arrow——一个跨语言的列式内存格式。

行式存储（Pandas/NumPy）          列式存储（Polars/Arrow）
┌──────────────────────┐     ┌──────────────────────┐
│ Row 0: [A, 25, 5000] │     │ name:  [A, B, C, D]  │
│ Row 1: [B, 30, 6000] │     │ age:   [25, 30, 35]  │
│ Row 2: [C, 35, 7000] │     │ salary:[5000,6000,7k] │
│ Row 3: [D, 40, 8000] │     └──────────────────────┘
└──────────────────────┘     每列连续存储，CPU缓存友好
每行跨列存储，缓存不友好

列式存储的性能优势：

1. CPU 缓存友好：连续内存访问，缓存命中率提升 5-10 倍
2. 向量化计算：单列数据可以用 SIMD 指令批量处理
3. 按需读取：只加载需要的列，不加载整行数据
4. 压缩效率高：同类型数据连续存储，压缩比可达 5-10 倍

import polars as pl

# 只读取需要的列——1GB 文件只加载 100MB
df = pl.scan_csv('large_dataset.csv').select(['id', 'value']).collect()

# Pandas 必须加载所有列再选择
df_pd = pd.read_csv('large_dataset.csv', usecols=['id', 'value'])

2.2 惰性计算引擎（Lazy Evaluation）

这是 Polars 最核心的性能优化。惰性计算让 Polars 能够全局优化查询计划，而不是逐行执行。

# Pandas：急切执行，每步都立即计算
df_pandas = pd.read_csv('data.csv')                    # ① 加载全部数据
result = df_pandas[df_pandas['age'] > 25]              # ② 过滤（创建副本）
result = result.groupby('city')['salary'].mean()        # ③ 聚合

# Polars：惰性执行，全局优化
df_polars = pl.scan_csv('data.csv')                    # ① 不加载数据，只记录操作
result = (df_polars
    .filter(pl.col('age') > 25)                         # ② 记录过滤（不执行）
    .groupby('city')                                    # ③ 记录聚合（不执行）
    .agg(pl.col('salary').mean())                       # ④ 记录聚合函数
    .collect()                                          # ⑤ 统一执行！
)

查询优化器做了什么？

原始查询计划：
  ① 读取 CSV（所有列，1000 万行）
  ② 过滤 age > 25
  ③ 按 city 分组
  ④ 计算 salary 均值

优化后查询计划：
  ① 读取 CSV（只读 age、city、salary 三列）← 列裁剪
  ② 过滤 age > 25（在读取时就过滤）← 谓词下推
  ③ 按 city 分组
  ④ 计算 salary 均值

实际执行：
  只加载 3 列 → 过滤后只剩 600 万行 → 聚合
  内存使用减少 70%，速度提升 3-5 倍

2.3 多线程并行策略

Polars 的 Rust 底层自动将任务分配到多个 CPU 核心：

# Polars 自动并行化
# 假设你有 8 个 CPU 核心
import polars as pl

df = pl.DataFrame({
    'group': np.random.choice(['A', 'B', 'C', 'D'], 10_000_000),
    'value': np.random.randn(10_000_000),
})

# groupby 自动并行——每个核心处理一个分组
result = df.groupby('group').agg([
    pl.col('value').mean(),
    pl.col('value').sum(),
    pl.col('value').std(),
])
# 8 核心并行：理论加速 8x，实际约 5-6x（有调度开销）

并行策略详解：

┌─────────────────────────────────────────────────┐
│                 Polars 线程池                     │
│                                                  │
│  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐        │
│  │Core 0│  │Core 1│  │Core 2│  │Core 3│        │
│  │Group │  │Group │  │Group │  │Group │        │
│  │  A   │  │  B   │  │  C   │  │  D   │        │
│  └──┬───┘  └──┬───┘  └──┬───┘  └──┬───┘        │
│     │         │         │         │             │
│  ┌──▼─────────▼─────────▼─────────▼───┐        │
│  │         结果合并（Reduce）            │        │
│  └────────────────────────────────────┘        │
│                                                  │
│  ┌────────────────────────────────────┐        │
│  │  工作窃取（Work Stealing）           │        │
│  │  空闲核心自动接管其他核心的待处理任务  │        │
│  └────────────────────────────────────┘        │
└─────────────────────────────────────────────────┘

2.4 表达式 API：Polars 的核心语言

Polars 的表达式 API 是其最独特的设计——所有操作都通过 pl.col() + 方法链完成：

import polars as pl

# 表达式组合：一个 select 中完成多个操作
df.select([
    # 基本选择
    pl.col('name'),
    
    # 数学运算
    pl.col('price') * pl.col('quantity').alias('total'),
    
    # 条件表达式
    pl.when(pl.col('age') >= 18)
      .then(pl.lit('adult'))
      .otherwise(pl.lit('minor'))
      .alias('age_group'),
    
    # 窗口函数（不改变行数）
    pl.col('salary').rank().over('department').alias('dept_rank'),
    
    # 字符串操作
    pl.col('email').str.extract(r'@(.+)', 1).alias('domain'),
    
    # 时间操作
    pl.col('timestamp').dt.hour().alias('hour'),
    
    # 聚合 + 命名
    pl.col('amount').sum().alias('total_amount'),
])

三、性能实战：Pandas vs Polars 基准测试

3.1 测试环境

• CPU: Apple M3 Pro（12 核）
• 内存: 36GB
• Python: 3.14
• Pandas: 2.3.0
• Polars: 1.28.0

3.2 数据生成

import polars as pl
import pandas as pd
import numpy as np

# 生成 1000 万行测试数据
N = 10_000_000

data = {
    'id': range(N),
    'user_id': np.random.randint(1, 100_000, N),
    'amount': np.random.randn(N) * 100 + 500,
    'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], N),
    'region': np.random.choice(['North', 'South', 'East', 'West'], N),
    'date': pd.date_range('2024-01-01', periods=N, freq='s'),
    'is_active': np.random.choice([True, False], N),
}

3.3 基准测试结果

import time

def benchmark(name, func, iterations=5):
    times = []
    for _ in range(iterations):
        start = time.time()
        func()
        times.append(time.time() - start)
    avg = sum(times) / len(times)
    print(f"{name}: {avg:.3f}s (avg of {iterations})")
    return avg

# 测试 1: 过滤 + 聚合
t1_pandas = benchmark("Pandas  过滤+聚合", lambda:
    pd.DataFrame(data).query('amount > 500').groupby('category')['amount'].mean()
)

t1_polars = benchmark("Polars  过滤+聚合", lambda:
    pl.DataFrame(data).filter(pl.col('amount') > 500)
        .groupby('category').agg(pl.col('amount').mean())
)

# 测试 2: 多列聚合
t2_pandas = benchmark("Pandas  多列聚合", lambda:
    pd.DataFrame(data).groupby(['category', 'region']).agg({
        'amount': ['mean', 'sum', 'std', 'count'],
        'user_id': 'nunique'
    })
)

t2_polars = benchmark("Polars  多列聚合", lambda:
    pl.DataFrame(data).groupby(['category', 'region']).agg([
        pl.col('amount').mean().alias('amount_mean'),
        pl.col('amount').sum().alias('amount_sum'),
        pl.col('amount').std().alias('amount_std'),
        pl.col('amount').count().alias('amount_count'),
        pl.col('user_id').n_unique().alias('user_nunique'),
    ])
)

# 测试 3: Join 操作
df_left = pd.DataFrame({'id': range(N), 'key': np.random.randint(1, 100_000, N)})
df_right = pd.DataFrame({'key': range(100_000), 'value': np.random.randn(100_000)})

t3_pandas = benchmark("Pandas  Join", lambda:
    df_left.merge(df_right, on='key', how='left')
)

lf_left = pl.DataFrame(df_left).lazy()
lf_right = pl.DataFrame(df_right).lazy()

t3_polars = benchmark("Polars  Join", lambda:
    lf_left.join(lf_right, on='key', how='left').collect()
)

结果汇总：

3.4 内存使用对比

import psutil
import os

def measure_memory(func):
    """测量函数执行期间的峰值内存使用"""
    process = psutil.Process(os.getpid())
    mem_before = process.memory_info().rss / 1e9
    func()
    mem_after = process.memory_info().rss / 1e9
    return mem_after - mem_before

# 加载 1GB CSV 文件的内存对比
pandas_mem = measure_memory(lambda: pd.read_csv('1gb_data.csv'))
polars_mem = measure_memory(lambda: pl.read_csv('1gb_data.csv'))

print(f"Pandas 内存增量: {pandas_mem:.2f} GB")
print(f"Polars 内存增量: {polars_mem:.2f} GB")
# 典型输出：
# Pandas 内存增量: 3.12 GB
# Polars 内存增量: 1.08 GB
# Polars 节省 65% 内存

四、从 Pandas 迁移到 Polars：完整实战

4.1 API 对照表

4.2 实战：电商数据分析全流程

import polars as pl

# ===== 1. 数据加载 =====
# 使用惰性模式——Polars 会在 collect() 时全局优化
orders = pl.scan_csv('orders.csv')
users = pl.scan_csv('users.csv')
products = pl.scan_csv('products.csv')

# ===== 2. 数据清洗 =====
clean_orders = (
    orders
    # 过滤无效订单
    .filter(
        pl.col('amount') > 0,
        pl.col('status') != 'cancelled',
    )
    # 处理时间列
    .with_columns([
        pl.col('created_at').str.strptime(pl.Datetime, '%Y-%m-%d %H:%M:%S'),
        pl.col('created_at').dt.date().alias('order_date'),
        pl.col('created_at').dt.hour().alias('order_hour'),
        pl.col('created_at').dt.month().alias('order_month'),
    ])
    # 金额分类
    .with_columns(
        pl.when(pl.col('amount') < 100).then(pl.lit('small'))
        .when(pl.col('amount') < 500).then(pl.lit('medium'))
        .otherwise(pl.lit('large'))
        .alias('amount_tier')
    )
)

# ===== 3. 关联查询 =====
# 用户订单关联
user_orders = (
    clean_orders
    .join(users, left_on='user_id', right_on='id', how='left')
    .join(products, left_on='product_id', right_on='id', how='left')
)

# ===== 4. 分析查询 =====

# 查询 1: 月度销售额趋势
monthly_sales = (
    user_orders
    .groupby('order_month')
    .agg([
        pl.col('amount').sum().alias('total_sales'),
        pl.col('order_id').n_unique().alias('order_count'),
        pl.col('user_id').n_unique().alias('unique_users'),
        pl.col('amount').mean().alias('avg_order_value'),
    ])
    .sort('order_month')
    .collect()
)

# 查询 2: 用户分层（RFM 分析）
rfm = (
    user_orders
    .groupby('user_id')
    .agg([
        # Recency: 最近一次购买距今天数
        (pl.col('order_date').max() - pl.col('order_date').min())
            .dt.total_days().alias('recency'),
        # Frequency: 购买次数
        pl.col('order_id').n_unique().alias('frequency'),
        # Monetary: 总消费金额
        pl.col('amount').sum().alias('monetary'),
    ])
    .with_columns([
        # 分层评分
        pl.col('recency').rank(reverse=True).alias('r_score'),
        pl.col('frequency').rank().alias('f_score'),
        pl.col('monetary').rank().alias('m_score'),
    ])
    .with_columns(
        (pl.col('r_score') + pl.col('f_score') + pl.col('m_score'))
            .alias('rfm_score')
    )
    .sort('rfm_score', descending=True)
    .collect()
)

# 查询 3: 商品类别交叉分析
category_cross = (
    user_orders
    .groupby(['category', 'amount_tier'])
    .agg([
        pl.col('amount').sum().alias('total'),
        pl.col('order_id').count().alias('count'),
    ])
    .pivot(
        values='total',
        index='category',
        columns='amount_tier',
        aggregate_function='sum',
    )
    .fill_null(0)
    .collect()
)

# 查询 4: 滑动窗口分析（7 天移动平均）
daily_sales = (
    user_orders
    .groupby('order_date')
    .agg(pl.col('amount').sum().alias('daily_total'))
    .sort('order_date')
    .with_columns(
        pl.col('daily_total')
            .rolling_mean(window_size=7)
            .alias('ma_7day')
    )
    .collect()
)

4.3 性能优化技巧

技巧 1：尽量使用 Lazy 模式

# ❌ 急切模式：每步都执行
df = pl.read_csv('big_file.csv')              # 立即加载
df = df.filter(pl.col('age') > 25)             # 立即过滤
df = df.select(['name', 'salary'])             # 立即选择列

# ✅ 惰性模式：全局优化后一次执行
df = (
    pl.scan_csv('big_file.csv')                # 不加载
    .filter(pl.col('age') > 25)                # 不执行
    .select(['name', 'salary'])                # 不执行
    .collect()                                  # 统一执行，自动优化
)

技巧 2：利用谓词下推减少 I/O

# Polars 的查询优化器会自动将过滤条件推到扫描阶段
# 等价于 SQL 的 WHERE 下推

# 这个查询：
pl.scan_csv('huge_file.csv')
  .filter(pl.col('year') == 2026)
  .select(['name', 'salary'])
  .collect()

# 优化器会变成：
# 1. 读取 CSV 时只加载 year、name、salary 三列（列裁剪）
# 2. 读取每行时检查 year == 2026，不符合的跳过（谓词下推）
# 3. 内存中只保留符合条件的行和列

技巧 3：使用 streaming 模式处理超大数据

# 当数据量超过内存时，使用 streaming 模式
result = (
    pl.scan_csv('100gb_file.csv')
    .filter(pl.col('amount') > 100)
    .groupby('category')
    .agg(pl.col('amount').sum())
    .collect(streaming=True)  # ← 关键参数
)
# streaming 模式会分块处理数据，内存使用保持稳定

技巧 4：避免 Python UDF，使用原生表达式

# ❌ 慢：Python UDF 打破并行和优化
df.with_columns(
    pl.col('text').map_elements(lambda x: x.upper())  # 逐行调用 Python
)

# ✅ 快：原生表达式
df.with_columns(
    pl.col('text').str.to_uppercase()  # Rust 原生实现，自动并行
)

五、Polars 的高级特性

5.1 嵌套数据类型

# Polars 原生支持 Struct 和 List 类型
df = pl.DataFrame({
    'name': ['Alice', 'Bob'],
    'scores': [[85, 90, 78], [92, 88, 95]],
    'address': [
        {'city': 'Beijing', 'zip': '100000'},
        {'city': 'Shanghai', 'zip': '200000'},
    ]
})

# 操作嵌套字段
df.select([
    pl.col('name'),
    pl.col('scores').list.mean().alias('avg_score'),
    pl.col('scores').list.max().alias('max_score'),
    pl.col('address').struct.field('city').alias('city'),
])

5.2 插件系统

# Polars 支持第三方表达式插件
# 安装: pip install polars-ols

import polars_ols  # 注册自定义表达式

df.with_columns(
    pl.col('y').ols.rolling_beta('x', window_size=30).alias('beta')
)

5.3 与 DuckDB 的互操作

# Polars DataFrame 可以直接传给 DuckDB
import duckdb

df = pl.DataFrame({'x': range(100), 'y': range(100)})

# 在 DuckDB 中查询 Polars DataFrame
result = duckdb.sql("SELECT SUM(x), AVG(y) FROM df").pl()
# 返回结果自动转换为 Polars DataFrame

5.4 Rust 原生 API

对于极致性能场景，可以直接使用 Polars 的 Rust API：

use polars::prelude::*;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 读取 CSV
    let df = CsvReadOptions::default()
        .with_has_header(true)
        .try_into_reader_with_file_path(Some("data.csv".into()))?
        .finish()?;
    
    // 惰性查询
    let result = df
        .lazy()
        .filter(col("age").gt(lit(25)))
        .groupby([col("city")])
        .agg([col("salary").mean().alias("avg_salary")])
        .sort(["avg_salary"], SortMultipleOptions::default().with_order_descending(true))
        .collect()?;
    
    println!("{:?}", result);
    Ok(())
}

六、Pandas vs Polars：迁移决策矩阵

6.1 何时选择 Polars

6.2 何时保留 Pandas

6.3 渐进式迁移策略

# 第一阶段：混合使用
# I/O 和大数据处理用 Polars，分析用 Pandas
df_polars = pl.scan_csv('huge_data.csv').filter(...).collect()
df_pandas = df_polars.to_pandas()  # 转换为 Pandas 供下游使用

# 第二阶段：核心管道迁移
# ETL 管道全部用 Polars，只在最终输出转 Pandas
def etl_pipeline():
    return (
        pl.scan_csv('raw/*.csv')
        .filter(...)
        .join(...)
        .groupby(...)
        .agg(...)
        .collect()
    )

# 第三阶段：全面迁移
# 包括特征工程、模型训练数据准备全部用 Polars

七、总结

7.1 Polars 的核心优势

1. 性能碾压：10-100 倍的速度优势，特别是大数据场景
2. 内存友好：列式存储 + 惰性计算节省 60-70% 内存
3. 类型安全：强类型系统避免 Pandas 的类型陷阱
4. 并行自动：无需手动配置，Rust 底层自动多线程
5. 查询优化：惰性计算引擎全局优化执行计划

7.2 Polars 的不足

1. 生态尚不成熟：相比 Pandas 的 15 年积累，第三方库支持有限
2. 时序功能弱：时间序列分析不如 Pandas 丰富
3. 学习曲线：表达式 API 与 Pandas 差异较大，需要适应
4. 可视化缺失：没有内建的可视化支持
5. 社区规模：Stack Overflow 上的问答数量远少于 Pandas

7.3 未来展望

Polars 正在快速发展，2026 年的路线图包括：

• GPU 加速：基于 CUDA 的表达式执行引擎
• 分布式计算：多节点数据并行
• SQL 接口：原生 SQL 查询支持
• 更多数据源：MongoDB、Elasticsearch 连接器
• AI 集成：与 scikit-learn、XGBoost 的原生对接

我的判断：Polars 是数据分析领域的"Rust 时刻"——不是立刻替代 Pandas，而是在性能敏感场景中逐步蚕食。就像 Rust 没有替代 C++ 但在系统编程领域建立了新标准，Polars 也将在大数据分析领域成为新的性能基准。2026 年，如果你还在用 Pandas 处理 GB 级数据，是时候认真看看 Polars 了。

参考资源：

• Polars 官方文档：https://pola-rs.github.io/polars/
• Polars GitHub：https://github.com/pola-rs/polars
• Polars vs Pandas 基准测试：https://www.pola.rs/benchmarks
• Apache Arrow 文档：https://arrow.apache.org/
• Polars 用户指南：https://pola-rs.github.io/polars-book/