polars

Polars

概述

Polars 是一款基于 Apache Arrow 构建、速度极快的 Python 和 Rust DataFrame 库。使用 Polars 基于表达式的 API、惰性求值框架和高性能数据处理能力，高效地处理数据、迁移 pandas 并优化数据管道。

快速入门

安装和基本用法

安装本次更新期间验证的当前稳定版 Polars：

uv pip install "polars==1.41.2"

仅在需要时安装可选集成：

uv pip install "polars[excel,database,fsspec,pandas,numpy]==1.41.2"

创建和操作 DataFrame 的基本方法：

import polars as pl

# Create DataFrame
df = pl.DataFrame({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "city": ["NY", "LA", "SF"]
})

# Select columns
df.select("name", "age")

# Filter rows
df.filter(pl.col("age") > 25)

# Add computed columns
df.with_columns(
    age_plus_10=pl.col("age") + 10
)

核心概念

表达式

表达式是 Polars 操作的基本构建块。它们描述对数据进行的转换，并且可以组合、复用和优化。

关键原则：

• 使用 pl.col("column_name") 引用列
• 链式调用方法，构建复杂转换
• 表达式是惰性的，仅在上下文中（select、with_columns、filter、group_by）执行

示例：

# Expression-based computation
df.select(
    pl.col("name"),
    (pl.col("age") * 12).alias("age_in_months")
)

惰性求值与即时求值

即时求值（DataFrame）： 操作立即执行

df = pl.read_csv("file.csv")  # Reads immediately
result = df.filter(pl.col("age") > 25)  # Executes immediately

惰性求值（LazyFrame）： 操作会构建查询计划，并在执行前进行优化

lf = pl.scan_csv("file.csv")  # Doesn't read yet
result = lf.filter(pl.col("age") > 25).select("name", "age")
df = result.collect()  # Now executes optimized query

适用场景：

• 处理大型数据集
• 复杂的查询管道
• 只需要部分列或行时
• 性能至关重要时

惰性求值的优势：

• 自动查询优化
• 谓词下推
• 投影下推
• 并行执行

如需了解详细概念，请加载 references/core_concepts.md。

常见操作

选择

选择和处理列：

# Select specific columns
df.select("name", "age")

# Select with expressions
df.select(
    pl.col("name"),
    (pl.col("age") * 2).alias("double_age")
)

# Select all columns matching a pattern
df.select(pl.col("^.*_id$"))

筛选

按条件筛选行：

# Single condition
df.filter(pl.col("age") > 25)

# Multiple conditions (cleaner than using &)
df.filter(
    pl.col("age") > 25,
    pl.col("city") == "NY"
)

# Complex conditions
df.filter(
    (pl.col("age") > 25) | (pl.col("city") == "LA")
)

添加列

添加或修改列，同时保留现有列：

# Add new columns
df.with_columns(
    age_plus_10=pl.col("age") + 10,
    name_upper=pl.col("name").str.to_uppercase()
)

# Parallel computation (all columns computed in parallel)
df.with_columns(
    pl.col("value") * 10,
    pl.col("value") * 100,
)

分组和聚合

对数据分组并计算聚合结果：

# Basic grouping
df.group_by("city").agg(
    pl.col("age").mean().alias("avg_age"),
    pl.len().alias("count")
)

# Multiple group keys
df.group_by("city", "department").agg(
    pl.col("salary").sum()
)

# Conditional aggregations
df.group_by("city").agg(
    (pl.col("age") > 30).sum().alias("over_30")
)

如需了解详细的操作模式，请加载 references/operations.md。

聚合和窗口函数

聚合函数

在 group_by 上下文中常用的聚合函数：

• pl.len() - 统计行数
• pl.col("x").sum() - 对值求和
• pl.col("x").mean() - 求平均值
• pl.col("x").min() / pl.col("x").max() - 求最小值 / 最大值
• pl.first() / pl.last() - 获取第一个 / 最后一个值

使用 over() 的窗口函数

应用聚合，同时保留行数：

# Add group statistics to each row
df.with_columns(
    avg_age_by_city=pl.col("age").mean().over("city"),
    rank_in_city=pl.col("salary").rank().over("city")
)

# Multiple grouping columns
df.with_columns(
    group_avg=pl.col("value").mean().over("category", "region")
)

映射策略：

• group_to_rows（默认）：保留原始行顺序
• explode：速度更快，但会将同组的行放在一起
• join：创建列表列

数据输入 / 输出

支持的格式

Polars 支持读取和写入：

• CSV、Parquet、JSON、Excel
• 数据库（通过连接器）
• 云存储（S3、Azure、GCS）
• Google BigQuery
• 多个文件 / 分区文件

常见输入 / 输出操作

CSV：

# Eager
df = pl.read_csv("file.csv")
df.write_csv("output.csv")

# Lazy (preferred for large files)
lf = pl.scan_csv("file.csv")
result = lf.filter(...).select(...).collect()

Parquet（推荐用于提升性能）：

df = pl.read_parquet("file.parquet")
df.write_parquet("output.parquet")

JSON：

df = pl.read_json("file.json")
df.write_json("output.json")

如需查看完整的输入 / 输出文档，请加载 references/io_guide.md。

转换

连接

合并 DataFrame：

# Inner join
df1.join(df2, on="id", how="inner")

# Left join
df1.join(df2, on="id", how="left")

# Join on different column names
df1.join(df2, left_on="user_id", right_on="id")

拼接

堆叠 DataFrame：

# Vertical (stack rows)
pl.concat([df1, df2], how="vertical")

# Horizontal (add columns)
pl.concat([df1, df2], how="horizontal")

# Diagonal (union with different schemas)
pl.concat([df1, df2], how="diagonal")

透视和逆透视

重塑数据：

# Pivot (wide format)
df.pivot(on="product", values="sales", index="date")

# Unpivot (long format)
df.unpivot(index="id", on=["col1", "col2"])

如需查看详细的转换示例，请加载 references/transformations.md。

pandas 迁移

与 pandas 相比，Polars 性能提升显著，且 API 更简洁。主要区别：

概念差异

• 没有索引：Polars 仅使用整数位置
• 严格类型：不会静默转换类型
• 惰性求值：可通过 LazyFrame 使用
• 默认并行：操作会自动并行执行

常见操作对照

操作	Pandas	Polars
选择列	df["col"]	df.select("col")
筛选	df[df["col"] > 10]	df.filter(pl.col("col") > 10)
添加列	df.assign(x=...)	df.with_columns(x=...)
分组	df.groupby("col").agg(...)	df.group_by("col").agg(...)
窗口	df.groupby("col").transform(...)	df.with_columns(...).over("col")

关键语法模式

Pandas 顺序执行（慢）：

df.assign(
    col_a=lambda df_: df_.value * 10,
    col_b=lambda df_: df_.value * 100
)

Polars 并行执行（快）：

df.with_columns(
    col_a=pl.col("value") * 10,
    col_b=pl.col("value") * 100,
)

如需完整的迁移指南，请加载 references/pandas_migration.md。

最佳实践

性能优化

1. 对大型数据集使用惰性求值：

   lf = pl.scan_csv("large.csv")  # Don't use read_csv
   result = lf.filter(...).select(...).collect()
   

2. 避免在热点路径中使用 Python 函数：

   • 使用表达式 API 进行并行化
   • 仅在必要时使用 .map_elements()
   • 优先使用 Polars 原生操作

3. 对超大型数据使用流式处理：

   lf.collect(engine="streaming")
   

4. 尽早仅选择所需列：

   # Good: Select columns early
   lf.select("col1", "col2").filter(...)
   
   # Bad: Filter on all columns first
   lf.filter(...).select("col1", "col2")
   

5. 使用适当的数据类型：

   • 对低基数字符串使用 Categorical
   • 使用适当的整数大小（i32 与 i64）
   • 对时间数据使用日期类型

表达式模式

条件操作：

pl.when(condition).then(value).otherwise(other_value)

跨多个列的列操作：

df.select(pl.col("^.*_value$") * 2)  # Regex pattern

空值处理：

pl.col("x").fill_null(0)
pl.col("x").is_null()
pl.col("x").drop_nulls()

如需了解更多最佳实践和模式，请加载 references/best_practices.md。

资源

此技能包含全面的参考文档：

references/

• core_concepts.md - 详细说明表达式、惰性求值和类型系统
• operations.md - 常见操作的完整指南及示例
• pandas_migration.md - 从 pandas 迁移到 Polars 的完整指南
• io_guide.md - 所有支持格式的数据 I/O 操作
• transformations.md - 联接、串联、透视和重塑操作
• best_practices.md - 性能优化技巧和常见模式

当用户需要了解特定主题的详细信息时，按需加载这些参考资料。

Agent 操作流程

1. 检查环境。 确认 Python 环境和库版本（python -c "import pkg; print(pkg.__version__)"），并检查数据的形状、类型和缺失值。
2. 明确输入。 根据数据或用户提供的信息确认格式、标识符和参数。对于任何会改变结果的值，应询问，而不是猜测。
3. 先运行小规模版本。 先在样本或单个折上运行，并检查运行时间和内存占用。
4. 使用上述说明和参考资料执行完整任务。
5. 验证结果。 使用留出数据、固定随机种子和适当的指标；检查数据泄漏；报告不确定性（置信区间、不同随机种子下的标准差）。
6. 报告结果。 说明运行了什么（版本、命令、参数）、检查了什么，以及仍有哪些不确定之处。

如果出现这种情况	请这样处理
内存不足或执行非常缓慢	对数据进行子采样，使用分块或惰性计算，或者缩小模型规模，并告知用户所做的更改。
这些说明中的函数、标志或端点在已安装版本中不存在	查阅已安装版本自带的文档（help()、--help、官方文档），进行适配，并告知用户。切勿编造 API。
必需的输入、标识符或参数不明确	询问用户，或在运行前明确说明所作的假设。

诚信规则

• 切勿捏造结果、参数、标识符、引文或统计数据。如果无法运行或验证某项内容，应如实说明。
• 切勿报告本次会话中未计算的指标；展示生成每个数值所经过的代码路径。
• 请将此处特定于版本的细节视为可能已过时：根据已安装版本的官方文档进行确认。
• 对于会产生费用、占用共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，必须先征求同意。

相关技能

• dask：用于处理规模超过内存容量的 pandas/NumPy 工作流的分布式计算。
• ray-data：用于机器学习工作负载的可扩展数据处理。
• vaex：使用 Vaex 处理和分析规模超过内存容量的表格数据集。Vaex 是一个 Python 库，可基于内存映射……