polars

Polars

概述

Polars 是一个用 Rust 编写并提供 Python 绑定的超高速 DataFrame 库。Polars 专为高性能和内存效率而设计，利用并行执行和惰性求值，比 pandas 更快地处理数据，尤其适用于大型数据集。

何时使用此技能

在用户出现以下情况时激活：

• 想使用 DataFrame，并需要高性能
• 明确提到 Polars，或要求进行“快速”数据处理
• 需要处理大型数据集（数百万行）
• 想使用惰性求值来优化查询
• 询问数据转换、筛选、分组或聚合
• 需要读取/写入 CSV、Parquet、JSON 或其他数据格式
• 想将 DuckDB 与 Polars 结合，用于 SQL + DataFrame 工作流

安装

检查是否已安装 Polars：

python3 -c "import polars as pl; print(pl.__version__)"

如果尚未安装：

pip3 install polars

要使用包括 Parquet 支持在内的完整功能：

pip3 install 'polars[pyarrow]'

要集成 DuckDB：

pip3 install polars duckdb 'polars[pyarrow]'

核心功能

1. 读取数据

Polars 可以读取多种格式的数据：

import polars as pl
# Read CSV
df = pl.read_csv('data.csv')

# Read Parquet (fast, columnar format)
df = pl.read_parquet('data.parquet')

# Read JSON
df = pl.read_json('data.json')

# Read multiple files
df = pl.read_csv('data/*.csv')

# Read with lazy evaluation (doesn't load until needed)
lazy_df = pl.scan_csv('large_data.csv')
lazy_df = pl.scan_parquet('data/*.parquet')

2. 数据选择与筛选

import polars as pl
df = pl.read_csv('data.csv')

# Select columns
result = df.select(['name', 'age', 'city'])

# Select with expressions
result = df.select([
    pl.col('name'),
    pl.col('age'),
    pl.col('salary').alias('annual_salary')
])

# Filter rows
result = df.filter(pl.col('age') > 25)

# Multiple conditions
result = df.filter(
    (pl.col('age') > 25) &
    (pl.col('city') == 'NYC')
)

# Filter with string methods
result = df.filter(pl.col('name').str.contains('John'))

3. 转换与新增列

import polars as pl
df = pl.read_csv('sales.csv')

# Add new columns
result = df.with_columns([
    (pl.col('quantity') * pl.col('price')).alias('total'),
    pl.col('product').str.to_uppercase().alias('product_upper'),
    pl.when(pl.col('quantity') > 10)
        .then(pl.lit('bulk'))
        .otherwise(pl.lit('retail'))
        .alias('sale_type')
])

# Modify existing columns
result = df.with_columns([
    pl.col('price').round(2),
    pl.col('date').str.strptime(pl.Date, '%Y-%m-%d')
])

# Rename columns
result = df.rename({'old_name': 'new_name'})

4. 聚合与分组

import polars as pl
df = pl.read_csv('sales.csv')

# Group by and aggregate
result = df.group_by('category').agg([
    pl.col('sales').sum().alias('total_sales'),
    pl.col('sales').mean().alias('avg_sales'),
    pl.col('sales').count().alias('num_sales'),
    pl.col('product').n_unique().alias('unique_products')
])

# Multiple group by columns
result = df.group_by(['category', 'region']).agg([
    pl.col('revenue').sum(),
    pl.col('customer_id').n_unique().alias('unique_customers')
])

# Aggregation without grouping
stats = df.select([
    pl.col('sales').sum(),
    pl.col('sales').mean(),
    pl.col('sales').median(),
    pl.col('sales').std(),
    pl.col('sales').min(),
    pl.col('sales').max()
])

5. 排序与排名

import polars as pl

df = pl.read_csv('data.csv')

# Sort by single column
result = df.sort('age')

# Sort descending
result = df.sort('salary', descending=True)

# Sort by multiple columns
result = df.sort(['department', 'salary'], descending=[False, True])

# Add rank column
result = df.with_columns([
    pl.col('salary').rank(method='dense').over('department').alias('dept_rank')
])

6. 联接

import polars as pl

customers = pl.read_csv('customers.csv')
orders = pl.read_csv('orders.csv')

# Inner join
result = customers.join(orders, on='customer_id', how='inner')

# Left join
result = customers.join(orders, on='customer_id', how='left')

# Join on different column names
result = customers.join(
    orders,
    left_on='id',
    right_on='customer_id',
    how='inner'
)

# Join on multiple columns
result = df1.join(df2, on=['col1', 'col2'], how='inner')

7. 窗口函数

import polars as pl
df = pl.read_csv('sales.csv')

# Calculate running total
result = df.with_columns([
    pl.col('sales').cum_sum().over('region').alias('running_total')
])

# Calculate rolling average
result = df.with_columns([
    pl.col('sales').rolling_mean(window_size=7).alias('7_day_avg')
])

# Rank within groups
result = df.with_columns([
    pl.col('sales').rank().over('category').alias('category_rank')
])

# Lag and lead
result = df.with_columns([
    pl.col('sales').shift(1).over('product').alias('prev_sales'),
    pl.col('sales').shift(-1).over('product').alias('next_sales')
])

使用惰性求值提升性能

Polars 的惰性 API 会在执行前优化查询：

import polars as pl
# Start with lazy scan (doesn't load data yet)
lazy_df = (
    pl.scan_csv('large_data.csv')
    .filter(pl.col('date') >= '2024-01-01')
    .select(['customer_id', 'product', 'sales', 'date'])
    .group_by('customer_id')
    .agg([
        pl.col('sales').sum().alias('total_sales'),
        pl.col('product').n_unique().alias('unique_products')
    ])
    .filter(pl.col('total_sales') > 1000)
    .sort('total_sales', descending=True)
)

# Execute the optimized query
result = lazy_df.collect()

# Or get execution plan
print(lazy_df.explain())

常见模式

模式 1：ETL 流水线

import polars as pl
from datetime import datetime

# Extract and Transform
result = (
    pl.scan_csv('raw_data.csv')
    # Clean data
    .filter(
        (pl.col('amount') > 0) &
        (pl.col('quantity') > 0)
    )
    # Transform columns
    .with_columns([
        pl.col('date').str.strptime(pl.Date, '%Y-%m-%d'),
        pl.col('product').str.strip().str.to_uppercase(),
        (pl.col('quantity') * pl.col('amount')).alias('total'),
        pl.when(pl.col('quantity') > 10)
            .then(pl.lit('bulk'))
            .otherwise(pl.lit('retail'))
            .alias('order_type')
    ])
    # Aggregate
    .group_by(['date', 'product', 'order_type'])
    .agg([
        pl.col('total').sum().alias('daily_total'),
        pl.col('quantity').sum().alias('daily_quantity'),
        pl.count().alias('num_orders')
    ])
    .collect()
)

# Load (save results)
result.write_parquet('processed_data.parquet')

模式 2：数据探索

import polars as pl

df = pl.read_csv('data.csv')

# Quick overview
print(df.head())
print(df.describe())
print(df.schema)

# Column statistics
print(df.select([
    pl.col('age').min(),
    pl.col('age').max(),
    pl.col('age').mean(),
    pl.col('age').median(),
    pl.col('age').std()
]))

# Count nulls
print(df.null_count())

# Value counts
print(df['category'].value_counts())

# Unique values
print(df['status'].n_unique())

模式 3：与 DuckDB 结合使用

使用 Polars 加载数据，使用 DuckDB 进行 SQL 分析：

import polars as pl, duckdb
# Load data with Polars
df = pl.read_parquet('data/*.parquet')

# Use DuckDB for complex SQL
result = duckdb.sql("""
    SELECT
        category,
        DATE_TRUNC('month', date) as month,
        SUM(revenue) as monthly_revenue,
        COUNT(DISTINCT customer_id) as unique_customers
    FROM df
    WHERE date >= '2024-01-01'
    GROUP BY category, month
    ORDER BY month DESC, monthly_revenue DESC
""").pl()  # Convert back to Polars DataFrame

# Continue with Polars
final = result.with_columns([
    (pl.col('monthly_revenue') / pl.col('unique_customers')).alias('revenue_per_customer')
])

模式 4：写入数据

import polars as pl
df = pl.read_csv('data.csv')

# Write to CSV
df.write_csv('output.csv')

# Write to Parquet (recommended for large data)
df.write_parquet('output.parquet')

# Write to JSON
df.write_json('output.json')

# Write partitioned Parquet files
df.write_parquet('output/', partition_by='date')

表达式链式调用

Polars 使用强大的表达式语法：

import polars as pl
result = df.select([
    # String operations
    pl.col('name').str.to_lowercase().str.strip().alias('clean_name'),

    # Arithmetic
    (pl.col('price') * 1.1).round(2).alias('price_with_tax'),

    # Conditional logic
    pl.when(pl.col('age') < 18)
        .then(pl.lit('minor'))
        .when(pl.col('age') < 65)
        .then(pl.lit('adult'))
        .otherwise(pl.lit('senior'))
        .alias('age_group'),

    # Date operations
    pl.col('date').dt.year().alias('year'),
    pl.col('date').dt.month().alias('month'),

    # List operations
    pl.col('tags').list.len().alias('num_tags'),
])

性能提示

1. 使用惰性求值处理大型数据集——这能让 Polars 优化查询
2. 使用 Parquet 格式——它是列式压缩格式，速度远快于 CSV
3. 尽早筛选——在其他操作之前应用筛选条件
4. 避免逐行迭代——改用向量化操作
5. 使用表达式——效率高于 pandas 风格的操作

# Good: Lazy + filter early
result = (
    pl.scan_parquet('large.parquet')
    .filter(pl.col('date') >= '2024-01-01')  # Filter first
    .select(['col1', 'col2', 'col3'])  # Then select
    .collect()
)

# Less efficient: Eager loading
df = pl.read_parquet('large.parquet')
result = df.filter(pl.col('date') >= '2024-01-01').select(['col1', 'col2', 'col3'])

Polars 与 Pandas 的比较

主要区别：

• 不可变性：Polars DataFrame 不可变（操作会返回新的 DataFrame）
• 性能：Polars 通常比 pandas 快 5-10 倍
• 惰性求值：Polars 可以在执行前优化查询
• 表达式：Polars 使用表达式 API，而不是方法链式调用
• 并行处理：Polars 会自动并行执行操作

# Pandas style
df['new_col'] = df['col1'] * df['col2']

# Polars style
df = df.with_columns([
    (pl.col('col1') * pl.col('col2')).alias('new_col')
])

与 DuckDB 集成

要兼得两者的优势，可以将 Polars 和 DuckDB 结合使用：

import polars as pl, duckdb
# Polars: Fast data loading and transformation
df = (
    pl.scan_parquet('data/*.parquet')
    .filter(pl.col('active') == True)
    .collect()
)

# DuckDB: SQL analytics
result = duckdb.sql("""
    SELECT
        category,
        SUM(amount) as total,
        AVG(amount) as average
    FROM df
    GROUP BY category
""").pl()

如需了解更多 SQL 功能，请参阅 duckdb 技能；如需详细了解 Polars API，请参阅 references/api_reference.md 文件。

错误处理

import polars as pl
try:
    df = pl.read_csv('data.csv')
except FileNotFoundError:
    print("File not found")
except pl.exceptions.ComputeError as e:
    print(f"Polars compute error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

资源

• references/api_reference.md：详细的 Polars API 文档和示例
• 官方文档：https://docs.pola.rs/
• API 参考：https://docs.pola.rs/api/python/stable/reference/