Polars 性能碾压 Pandas 吗？千万级真实业务数据实测给出答案

硬件: 8 核 16G 服务器
Python: 3.12.1
Pandas: 2.2.0
Polars: 0.20.15
数据量: 1000 万行 × 12 列（CSV 文件 567 MB）

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

np.random.seed(42)
N = 10_000_000

df_source = pd.DataFrame({
    'order_id':   range(N),
    'user_id':    np.random.randint(1, 500_000, N),
    'product_id': np.random.randint(1, 10_000, N),
    'amount':     np.random.uniform(10, 5000, N).round(2),
    'quantity':   np.random.randint(1, 50, N),
    'region':     np.random.choice(['华东', '华南', '华北', '西南', '东北'], N),
    'category':   np.random.choice(['电子', '服装', '食品', '家居', '图书'], N),
    'status':     np.random.choice(['paid', 'pending', 'refunded', 'shipped'], N),
    'channel':    np.random.choice(['app', 'web', 'mini_program'], N),
    'order_date': pd.date_range('2023-01-01', periods=N, freq='s'),
})

df_source.to_csv('orders_10m.csv', index=False)

def benchmark(func, *args, repeat=3, **kwargs):
    """执行多次取最优值，减少系统调度波动"""
    times = []
    for _ in range(repeat):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        times.append(time.perf_counter() - start)
    return min(times), result

# Pandas
def pandas_read():
    return pd.read_csv('orders_10m.csv',
        dtype={'region': 'category', 'category': 'category', 'status': 'category'},
        parse_dates=['order_date'])

# Polars（默认多线程解析）
def polars_read():
    return pl.read_csv('orders_10m.csv',
        schema_overrides={'order_date': pl.Datetime},
        try_parse_dates=True)

p_time, _ = benchmark(pandas_read)
pl_time, _ = benchmark(polars_read)
print(f"Pandas: {p_time:.2f}s | Polars: {pl_time:.2f}s | 加速比: {p_time/pl_time:.1f}x")

# Pandas：筛选金额 >1000 且华东地区的已付款订单
def pandas_filter(df):
    return df[(df['amount'] > 1000) & (df['region'] == '华东') & (df['status'] == 'paid')]

# Polars Eager
def polars_filter(df):
    return df.filter(
        (pl.col('amount') > 1000) &
        (pl.col('region') == '华东') &
        (pl.col('status') == 'paid'))

# Polars Lazy（惰性执行，可优化查询计划）
def polars_lazy_filter(lf):
    return lf.filter(
        (pl.col('amount') > 1000) &
        (pl.col('region') == '华东') &
        (pl.col('status') == 'paid')).collect()

# Pandas
def pandas_groupby(df):
    return df.groupby(['region', 'category']).agg(
        total_amount=('amount', 'sum'),
        avg_amount=('amount', 'mean'),
        max_amount=('amount', 'max'),
        order_count=('order_id', 'count'),
        unique_users=('user_id', 'nunique'),
    ).reset_index()

# Polars
def polars_groupby(df):
    return df.group_by(['region', 'category']).agg([
        pl.col('amount').sum().alias('total_amount'),
        pl.col('amount').mean().alias('avg_amount'),
        pl.col('amount').max().alias('max_amount'),
        pl.col('order_id').count().alias('order_count'),
        pl.col('user_id').n_unique().alias('unique_users'),
    ])

import requests
import polars as pl
from io import StringIO
# 通过亿牛云 API 代理获取出口 IP，避免单 IP 被限流
api_url = "http://ip.16yun.cn:817/myip/pl/<ORDER_ID>/?s=<SIGN>&u=<USER>&format=json"
proxy_info = requests.get(api_url, timeout=10).json()[0]
proxy = f"http://{proxy_info['ip']}:{proxy_info['port']}"
# 采集数据（返回 403 检查白名单，返回 429 降低频率）
resp = requests.get("https://api.example.com/export",
                    proxies={"http": proxy, "https": proxy}, timeout=30)
# 直接从内存字符串载入 Polars，跳过文件落盘
df = pl.read_csv(StringIO(resp.text), try_parse_dates=True)
# Polars Lazy 链式分析
result = (df.lazy()
    .filter(pl.col('amount') > 100)
    .group_by('region')
    .agg(pl.col('amount').sum().alias('revenue'))
    .collect()
)