polars

Polars - 高性能数据框

Polars 专为速度而设计。与在单个 CPU 核心上按顺序处理数据的 pandas 不同，Polars 会在所有可用核心上并行执行操作。其“惰性 API”允许在执行前优化查询，从而大幅降低内存开销和处理时间。

适用场景

• 处理在 pandas 中难以应对的大型数据集（1GB - 100GB+）。
• 执行速度优先的场景（Polars 通常比 pandas 快 10-100 倍）。
• 处理复杂的数据转换流水线（惰性求值）。
• RAM 有限的系统（Polars 比 pandas 更节省内存）。
• 需要严格的类型安全和一致的空值处理的场景。
• 读取/写入大型 Parquet、CSV 或 Avro 文件。

参考文档

官方文档：https://docs.pola.rs/
用户指南：https://docs.pola.rs/user-guide/
搜索模式：pl.DataFrame、pl.LazyFrame、pl.col、df.select、df.filter、df.group_by

核心原则

即时 API 与惰性 API

• 即时：立即执行操作（类似 pandas）。
• 惰性：将操作加入查询计划。Polars 会优化该计划（例如，谓词下推、投影下推），并仅在调用时执行。

表达式 API

Polars 使用声明式语法。无需编写循环或复杂的 lambda，只需使用 pl.col() 编写表达式。这些表达式经过高度优化，并可并行运行。

Apache Arrow

Polars 以 Apache Arrow 格式存储数据，可与 PyArrow、DuckDB 等工具进行零拷贝数据交换。

快速参考

安装

pip install polars
# For Excel/Cloud support
pip install 'polars[all]'

标准导入

import polars as pl
import numpy as np

基本模式 - 惰性工作流（“Polars 方式”）

import polars as pl

# 1. Scan (Lazy) - doesn't load data yet
lf = pl.scan_csv("massive_data.csv")

# 2. Build Query Plan
query = (
    lf.filter(pl.col("age") > 25)
    .group_by("city")
    .agg([
        pl.col("salary").mean().alias("avg_salary"),
        pl.col("name").count().alias("count")
    ])
    .sort("avg_salary", descending=True)
)

# 3. Collect (Execute)
df = query.collect()

关键规则

✅ 应当做

• 优先使用惰性 API（scan_*） - 这样 Polars 就能优化内存使用并跳过不必要的数据。
• 使用表达式 - 始终使用 pl.col("name")，而不是通过字符串或索引选择列。
• 方法链式调用 - Polars 专为简洁、易读的流水线而构建。
• 指定 Schema - 读取 CSV 时，提供 Schema 可避免类型推断错误并加快加载速度。
• 使用 collect(streaming=True) - 对于大于 RAM 的数据集，流式处理可让 Polars 分块处理数据。
• 优先使用 Parquet 而非 CSV - 使用 Parquet 进行永久存储；它速度显著更快，并且会存储类型信息。

❌ 不要做

• 避免使用 .apply() - 自定义 Python 函数速度较慢，因为它们会破坏 Rust/并行优化。请使用内置表达式。
• 不要使用 inplace=True - Polars（与 JAX 类似）倾向于不可变性；转换会返回新的 DataFrame。
• 不要过早转换为 pandas - 尽可能长时间让数据保留在 Polars 中，以保持速度。
• 避免逐行迭代 - for row in df 是反模式；请使用向量化表达式。

反模式（绝对不要）

import polars as pl

# ❌ BAD: Using Python lambdas for simple math
# df.select(pl.col("val").map_elements(lambda x: x * 2)) # Slow!

# ✅ GOOD: Use expressions
df.select(pl.col("val") * 2) # Fast, parallelized in Rust

# ❌ BAD: Filtering after a heavy operation
# df.group_by("id").mean().filter(pl.col("id") == 5)

# ✅ GOOD: Lazy API will automatically "push down" the filter
(pl.scan_csv("data.csv")
 .filter(pl.col("id") == 5) # Optimized to read only id=5
 .group_by("id").mean())

# ❌ BAD: Converting to pandas just to check .head()
# df.to_pandas().head() 

# ✅ GOOD: Polars has its own fast .head() and rich printing
print(df.head())

表达式 API 深入解析

选择与转换

df.select([
    pl.col("name"),
    pl.col("price") * 1.2, # Scalar math
    pl.col("category").str.to_uppercase(), # String methods
    pl.col("date").dt.year().alias("year") # Date methods
])

筛选

# Multiple conditions
df.filter(
    (pl.col("price") < 100) & 
    (pl.col("status") == "active") |
    (pl.col("category").is_in(["A", "B"]))
)

聚合与分组

高性能统计

results = df.group_by("department").agg([
    pl.col("salary").sum(),
    pl.col("salary").max().alias("max_pay"),
    pl.col("name").n_unique().alias("unique_employees"),
    # Advanced: conditional aggregation inside group
    pl.col("salary").filter(pl.col("role") == "manager").mean().alias("manager_avg")
])

连接与拼接

类 SQL 操作

# Joins: 'inner', 'left', 'outer', 'semi', 'anti', 'cross'
df_joined = df_a.join(df_b, on="id", how="left")

# As-of join (for time-series alignment)
df_aligned = df_trades.join_asof(df_quotes, on="timestamp", by="symbol")

# Concatenation
df_stacked = pl.concat([df1, df2], how="vertical")

重塑（透视与熔化）

# Pivot
pivoted = df.pivot(values="sales", index="date", columns="region", aggregate_function="sum")

# Melt (Unpivot)
melted = df.melt(id_vars="date", value_vars=["store_a", "store_b"])

实用工作流

1. 大规模数据清理流水线

def clean_and_optimize(path):
    return (
        pl.scan_parquet(path)
        .drop_nulls(subset=["user_id"])
        .with_columns([
            pl.col("email").str.to_lowercase(),
            pl.col("timestamp").str.to_datetime("%Y-%m-%d %H:%M:%S"),
            (pl.col("income") / 1000).cast(pl.Float32) # Downcast for memory
        ])
        .filter(pl.col("timestamp") > pl.date(2023, 1, 1))
        .collect(streaming=True)
    )

2. 时间序列特征工程

def engineer_features(df):
    return df.with_columns([
        # Rolling average
        pl.col("price").rolling_mean(window_size="7d", by="date").alias("rolling_7d"),
        # Lead/Lag
        pl.col("price").shift(1).alias("prev_price"),
        # Cumulative sum
        pl.col("sales").cum_sum().over("category")
    ])

3. 快速解析 JSON/日志

def parse_logs(path):
    return (
        pl.scan_ndjson(path) # Read line-delimited JSON
        .select([
            "level",
            pl.col("message").str.extract(r"Error: (.*)", 1),
            pl.col("metadata").struct.field("user_id") # Access nested fields
        ])
        .collect()
    )

性能优化

with_columns 的优势

不要一次创建一列，而应使用 with_columns 并行执行多个计算。

# All 3 columns are calculated simultaneously in different threads
df = df.with_columns([
    (pl.col("a") + pl.col("b")).alias("sum"),
    (pl.col("a") * pl.col("b")).alias("prod"),
    pl.col("c").str.len().alias("c_len")
])

通过 Dtypes 选择列

快速对多组列应用转换。

# Multiply all float columns by 100
df = df.with_columns(
    pl.col(pl.Float64) * 100
)

常见问题与解决方案

.apply() 陷阱

.map_elements()（以前称为 .apply()）中的 Python 函数速度较慢。

# ❌ Problem: Using custom Python code
# df.select(pl.col("txt").map_elements(my_custom_func))

# ✅ Solution: Use Polars native expressions or pl.when()
df.select(
    pl.when(pl.col("score") > 50).then(pl.lit("Pass")).otherwise(pl.lit("Fail"))
)

大文件导致内存错误

如果 .collect() 遇到 OOM，可能是因为试图将过多数据加载到内存中。

# ✅ Solution: 
# 1. Use .filter() early in the Lazy plan.
# 2. Use streaming: .collect(streaming=True).
# 3. Select only the columns you need.

字符串与分类类型

对于低基数字符串（如“城市”或“性别”），请使用 Categorical。

# ✅ Solution: Saves massive amounts of RAM and speeds up joins
df = df.with_columns(pl.col("category").cast(pl.Categorical))

最佳实践

1. 处理大型文件时始终使用 Lazy API - 使用 scan_csv() 或 scan_parquet()，而不是 read_csv() 或 read_parquet()。
2. 收集数据前构建完整的查询计划 - 让 Polars 优化整个管道。
3. 优先使用表达式，而非 Python 函数 - 使用 pl.col() 表达式以获得最佳性能。
4. 读取 CSV 时指定模式 - 避免类型推断带来的开销和错误。
5. 处理超出内存容量的数据集时使用流式处理 - 在 collect() 中启用 streaming=True。
6. 优先使用 Parquet 格式 - 读写速度更快，并且能保留类型信息。
7. 将低基数字符串转换为 Categorical - 可显著降低内存占用并提升性能。
8. 使用 with_columns 进行多项转换 - 并行创建列。
9. 在惰性查询中尽早筛选 - 谓词下推可减少扫描的数据量。
10. 避免转换为 pandas - 留在 Polars 生态系统中，以获得最佳速度。

Polars 是单节点数据处理的新标杆。它结合了 Rust 的安全性与 Python 的灵活性，为现代数据科学带来流畅且极快的体验。