Polars 深度实战：当 Python 数据处理终于不再「等」——从 Arrow 内核、查询优化器到生产级迁移指南

前言

凌晨三点，你盯着一个跑了一小时还没出结果的 Jupyter Notebook——1000 万行数据，groupby().apply() 把内存吃到 32GB，然后 OOM 了。

这不是段子，这是 Pandas 在 2026 年的真实瓶颈。

Pandas 诞生于 2008 年，设计目标是「让金融数据分析更方便」。那时候 CPU 还是单核，数据还在内存里，CSV 文件撑死几十 MB。Pandas 完美解决了那个时代的问题。但 2026 年的数据环境完全不同：GB 级的 Parquet 文件、AI 训练的特征工程、实时流数据处理——Pandas 的单线程设计、GIL 全局锁、逐行迭代的运算是它骨子里的原罪，改不动了。

Polars 就是在这个问题上长出来的。它用 Rust 从头重写，内核没有一行 Python 代码，不受 GIL 限制，天然多线程，Arrow 列式内存格式，SIMD 向量化，惰性求值配合查询优化器——1000 万行 CSV 读取，Polars 比 Pandas 快 4-5 倍；GroupBy 聚合，快 5 倍；多表 Join，快 4 倍；内存占用只有 Pandas 的四分之一。

这不是「更快一点」，这是「重新定义什么叫快」。

本文从 Rust 内核设计哲学出发，深入拆解 Polars 的核心架构：Arrow 内存格式的零拷贝秘密、惰性求值的查询计划生成与优化机制、SIMD 向量化引擎的实现原理，再到表达式 API、窗口函数、时间序列处理等实战能力，最后给出从 Pandas 迁移的完整路径与避坑指南。目标：让你读完这篇文章后，能独立用 Polars 重写你团队里那个跑了一夜的 ETL 脚本。

一、背景：为什么 Pandas 会在 2026 年成为瓶颈

1.1 Pandas 的历史包袱

Pandas 的核心数据结构 DataFrame，底层依赖两个组件：NumPy ndarray 和 Python list of objects。

NumPy ndarray 是一块连续的内存区域，存放着相同类型的数值数据。当你做 df['a'] + df['b'] 时，NumPy 可以通过 SIMD 指令一次性对多个元素做并行运算——这一步本身是快的。

但问题出在其他地方：

第一，GIL（Global Interpreter Lock）。Python 的多线程是假的——同一时刻只有一个线程能执行 Python 字节码。你无法让多个 CPU 核心同时跑 Pandas 的操作，因为它们都在争抢同一个 GIL。要真正并行，只能走 multiprocessing，但进程间通信的序列化开销又抵消了并行的收益。

第二，逐行 Python 对象。当你读取一张包含字符串日期、混合类型的 CSV 时，Pandas 会把每一行的每个单元格都包装成一个 Python 对象（PyObject*）。这不是连续内存，而是一个链表。CPU 缓存完全失效，SIMD 指令用不了，整个运算退化成「在沙子里找金子」。

第三，没有查询优化器。当你写：

result = df[df['status'] == 'active']
        .groupby('region')
        .agg({'amount': 'sum'})

Pandas 收到这条指令后，是严格按照你写的顺序执行的。它先做过滤，再做分组聚合——这不是因为它「聪明」地做了优化，而是因为它根本没有优化器。你写了什么顺序，它就执行什么顺序，没有重排、没有下推、没有剪枝。

第四，inplace 参数的设计混乱。df.apply() 默认 inplace=False，返回新对象；但有些操作没有 inplace 参数，有些有却不生效。这种不一致性在数据量大时会变成灾难——每次复制 DataFrame 都是一次内存拷贝。

1.2 Polars 的设计哲学

Polars 的回答是：把数据处理的控制权从 Python 运行时还给编译器/硬件。

它的核心哲学有三条：

1. Rust-first：核心计算引擎完全用 Rust 编写，不经过 Python 解释器。Rust 没有 GIL，天然支持真正的多线程。

2. 内存即结构：所有数据使用 Apache Arrow 列式内存格式。列式存储让 SIMD 向量化成为可能，CPU 缓存命中率极高。

3. 先计划，再执行：惰性求值（Lazy Evaluation）。你写的每一条数据处理管道，先被翻译成一个逻辑查询计划，由优化器重排、重构，确认最优执行路径后，才真正执行。

这三条哲学，恰好分别解决了 Pandas 的四个历史包袱。

1.3 Arrow 列式内存格式：数据不搬家

要理解 Polars 为什么快，必须先理解 Apache Arrow。

Arrow 是一个列式内存格式规范，也是一个跨语言的标准化数据表示。它解决了数据科学中最浪费的一个问题：序列化/反序列化。

传统数据流是这样的：

CSV/Parquet → Pandas (Python 对象) → NumPy 数组 → Matplotlib → 存储
              ↑ 每次转换都要数据拷贝

Arrow 的列式存储是这样的：

[列1][列2][列3]...[列N]  ← 连续内存，CPU 直接访问

每一列是一段连续内存，相同类型数据紧密排列。读取时，CPU 可以预取（prefetch）整列数据进缓存；SIMD 指令一次处理多个值；GC（垃圾回收）的压力也小得多。

更重要的是 零拷贝。Polars 读取 Parquet 文件时，数据直接从磁盘映射到 Arrow 格式的内存区域，中间没有「转换」步骤——不需要把 Parquet 的压缩块解压成 Python 对象再转成 NumPy 数组。数据只存在一份，Polars 直接在上面做运算。

1.4 与 DuckDB 的关系：不是竞争，是协同

程序员茄子上有一篇讲 DuckDB 的文章（DuckDB 1.5 深度实战：当分析型数据库压缩进一个文件），讲的是「把 OLAP 查询引擎嵌入一个二进制文件」。

Polars 和 DuckDB 有重叠的能力——两者都擅长分析型查询，都用 Arrow，都很快。但它们的定位有本质区别：

在实际生产中，最佳实践是联合使用：用 Polars 做数据清洗和特征工程（Python 工作流），用 DuckDB 做复杂报表查询（SQL 工作流）。两者共享 Arrow 内存格式，Polars 输出的 DataFrame 可以直接被 DuckDB 查询，不需要序列化。

二、核心概念深度拆解

2.1 表达式 API：超越 SQL 的类型安全

Polars 的表达式 API 是我认为它设计得最优雅的部分。

在其他数据处理库里，你会看到这样的代码：

# Pandas - 操作分散，隐含假设多
df['total'] = df['price'] * df['qty']
df['grade'] = df['score'].apply(lambda x: 'A' if x >= 90 else 'B' if x >= 70 else 'C')

这里有三个问题：df['price'] 返回的是什么类型？apply 里的 lambda 怎么优化？df['score'] 是字符串还是数字？

Polars 的表达式 API 把这些问题全部消灭了：

# Polars - 所有操作都是表达式，类型安全，可优化
(
    df.lazy()
    .with_columns(
        (pl.col('price') * pl.col('qty')).alias('total'),
        pl.when(pl.col('score') >= 90)
          .then(pl.lit('A'))
          .when(pl.col('score') >= 70)
          .then(pl.lit('B'))
          .otherwise(pl.lit('C'))
          .alias('grade')
    )
    .filter(pl.col('status') == 'active')
    .group_by('region')
    .agg(
        pl.col('total').sum().alias('total_revenue'),
        pl.col('score').mean().alias('avg_score')
    )
    .sort('total_revenue', descending=True)
    .collect()
)

这条管道的每一段都是一个表达式，Polars 可以分析整个表达式树，决定最优执行顺序。

表达式的核心类型：

• pl.col('name') — 引用某一列
• pl.lit(value) — 字面量常量
• pl.when().then().otherwise() — 条件表达式
• pl.col('x').sum().over('partition') — 窗口函数（分组聚合后返回完整行）
• list() / struct() — 嵌套数据结构

2.2 惰性求值：Polars 的查询优化器

Polars 的惰性模式是它区别于 Pandas 的核心能力。当你调用 .lazy() 时，Polars 不会立即执行任何操作，而是：

第一步：构建逻辑查询计划（DAG）

query = (
    df.lazy()
    .filter(pl.col('amount') > 1000)
    .with_columns((pl.col('price') * pl.col('qty')).alias('revenue'))
    .group_by('region')
    .agg(pl.col('revenue').sum())
    .sort('revenue', descending=True)
)

这一步 Polars 构建了一个有向无环图（DAG），表示所有操作之间的依赖关系：

Filter(amount > 1000)
  └── WithColumns(revenue = price * qty)
        └── GroupBy(region)
              └── Agg(revenue_sum)
                    └── Sort(revenue DESC)

第二步：优化器重排（Query Optimization）

Polars 的优化器会对 DAG 做以下优化：

① 谓词下推（Predicate Pushdown）

原始顺序: WithColumns → Filter → GroupBy
优化后:   Filter → GroupBy → WithColumns

过滤条件先执行，减少后续操作处理的数据量。这是关系型数据库查询优化的标准技术，Polars 把这个能力带到了 DataFrame 层面。

② 列裁剪（Column Pruning）

如果 SELECT a, b FROM table WHERE c > 0，Polars 只读取列 a、b、c，其他列的磁盘 I/O 完全跳过。

③ 谓词折叠（Predicate Simplification）

如果写 filter((x > 5) & (x < 3))，Polars 识别出这是互斥条件，直接返回空结果，不做任何 I/O。

④ 广播优化（Broadcasting）

当小表 JOIN 大表时，Polars 把小表广播到每个工作线程，避免数据 shuffle。

第三步：执行计划可视化

# 打印优化后的物理执行计划
query.explain(optimized=True)

输出类似这样：

AGGREGATE
  col([sum(revenue)])
  BY(home_team)
SORT [sum(revenue) DESC]
  WITH_COLUMNS [(field_0 * field_1) AS revenue]
FILTER [(field_2) > (1000)]     ← 谓词下推成功
  CSV SCAN fields: [field_0, field_1, field_2]
  OPERATIONS: [selection, projection]

这样你可以清楚地看到优化器做了什么，以及哪些优化没有生效（可能是因为你的写法限制了优化空间）。

2.3 SIMD 向量化：让 CPU 满血跑数据

SIMD（Single Instruction Multiple Data）即单指令流多数据流。简单说就是：一条 CPU 指令同时处理多个数据。

现代 x86 CPU 的 AVX-512 指令集可以一次性处理 16 个 32 位整数或 8 个 64 位浮点数。如果你的数据存储在连续内存中，CPU 可以一次性把这 16 个整数从 L1 缓存读入寄存器，做一次加法，再写回——一条指令完成了 16 次运算。

Polars 的 Arrow 列式存储天然符合 SIMD 的条件：每列是连续内存，同一类型，CPU 预取器（prefetcher）可以提前把下一批数据读入缓存。Rust 编写的计算内核直接使用 SIMD 指令，不需要经过 Python 解释器的调度。

具体来说，Polars 在以下操作中大量使用 SIMD：

• 数值列的加减乘除、比较运算
• 字符串列的编码转换、正则匹配（利用 SIMD 字符串处理库如 simd-lite）
• 日期时间列的解析和计算
• 位运算和掩码操作

对比 Pandas：NumPy 也支持 SIMD，但受 GIL 限制，无法跨线程并行。在多核机器上，Polars 可以同时跑多个 SIMD 向量化运算，每个 CPU 核心处理数据的一个分区。

2.4 流式处理（Streaming）：GB 级数据不用全量内存

Polars 1.0 引入的流式处理引擎（Streaming Engine）是它工程能力的集中体现。

传统 DataFrame 的执行模式是：把所有数据加载到内存 → 执行所有操作 → 输出结果。这对 GB 级的数据来说意味着：你需要 2-3 倍数据大小的内存（一路管道下来，中间结果会膨胀），而且要等所有数据都就位才开始处理。

Polars 的流式处理改变了这一点：

# 流式执行：数据分批处理，边读边算，不吃满内存
result = (
    pl.scan_csv('huge_file.csv')        # scan_ 而不是 read_，流式读取
    .filter(pl.col('status') == 'completed')
    .group_by('customer_id')
    .agg(pl.col('amount').sum().alias('total'))
    .sink_parquet('output.parquet')     # 边算边写，不落盘到内存
)

scan_csv 不读取文件内容，只是建立一个查询计划；sink_parquet 不在内存中生成完整 DataFrame，而是把结果分批写入 Parquet 文件。整个过程只需要足够容纳一批数据的内存，而不是整个文件。

这在 ETL 场景下特别有价值：处理 50GB 的日志文件，不需要 50GB 内存，2GB 足够。

三、架构深度解析

3.1 多层架构：从用户代码到硬件指令

Polars 的架构分为五层，每一层都有明确职责：

┌─────────────────────────────────────┐
│  用户层 (Python / Node.js / Rust)   │  ← 表达 API，用户编写数据处理逻辑
├─────────────────────────────────────┤
│  IR 层 (Intermediate Representation)│  ← 查询计划，表达式树
├─────────────────────────────────────┤
│  优化器层 (Query Optimizer)          │  ← 谓词下推、列裁剪、剪枝
├─────────────────────────────────────┤
│  执行引擎 (Execution Engine)        │  ← 并行执行、流式处理
├─────────────────────────────────────┤
│  Rust 内核层 (Arrow / SIMD)          │  ← Arrow 内存管理、SIMD 指令
└─────────────────────────────────────┘

3.2 零拷贝数据访问：Arrow 的底层机制

Arrow 的零拷贝实现依赖两个关键技术：

① 内存映射（Memory Mapping）

Polars 读取 Parquet 文件时，使用 mmap（内存映射）系统调用：

# mmap 机制：文件内容不加载到内存，只建立虚拟地址映射
# 访问时触发缺页中断，OS 从磁盘读取对应数据块到内存
# 多进程共享同一个 mmap 区域，不额外占用内存
df = pl.read_parquet('data.parquet', memory_map=True)

多个 Polars 操作可以共享同一个内存映射区域，不产生额外拷贝。只有当某块数据被修改时，才会产生 Copy-on-Write。

② 公共子表达式消除（Common Subexpression Elimination）

在表达式 DAG 中，如果同一个列被多次引用（比如 df.with_columns(a+b, a*2)），Polars 只计算一次 a，然后共享给两个下游操作。

3.3 并行执行模型：真正的多线程

Polars 的并行执行有两种模式：

数据并行（Data Parallelism）：把 DataFrame 按行分区，每个 CPU 核心处理一个分区。这是 Join、Sort、GroupBy 等操作的默认并行方式。

# Polars 自动检测 CPU 核心数，启动对应数量的线程
# 线程数通过 n_ threads 配置，或 os.cpu_count() 决定
result = df.group_by('region').agg(pl.col('amount').sum())

# 查看/设置线程数
import polars as pl
print(pl.thread_pool_size())  # CPU 核心数
pl.Config.set_threadpool_size(8)

流水线并行（Pipeline Parallelism）：在流式模式下，Polars 让读取、计算、写入三条流水线并行运行，而不是等读取完全结束才开始计算。

四、代码实战：10 个生产级示例

4.1 环境准备与基础查询

import polars as pl
import numpy as np

# 基础读取
df = pl.read_csv('sales_data.csv')
print(df.head())
print(df.describe())

# 懒加载读取（推荐生产使用）
lf = pl.scan_csv('sales_data.csv')  # 不实际读取，建立计划

4.2 过滤、转换与聚合

# 场景：从销售数据中找出高价值客户
top_customers = (
    df.lazy()
    .filter(
        (pl.col('order_date') >= '2026-01-01') &
        (pl.col('status') == 'completed')
    )
    .with_columns(
        (pl.col('price') * pl.col('quantity')).alias('revenue'),
        (pl.col('order_date').dt.year()).alias('year'),
        (pl.col('order_date').dt.month()).alias('month'),
    )
    .group_by('customer_id')
    .agg([
        pl.col('revenue').sum().alias('total_revenue'),
        pl.col('revenue').mean().alias('avg_order_value'),
        pl.len().alias('order_count'),
        pl.col('product_category').n_unique().alias('category_diversity'),
    ])
    .filter(pl.col('order_count') >= 5)       # 至少 5 单
    .sort('total_revenue', descending=True)
    .limit(100)
    .collect()
)

print(top_customers)

4.3 窗口函数：跨行计算不破坏原表结构

窗口函数是数据分析中最强大的工具之一，Pandas 实现起来很繁琐，Polars 一行搞定：

# 计算每个区域占总销售额的百分比
result = (
    df.lazy()
    .with_columns(
        pl.col('revenue').sum().over('region').alias('region_total'),
    )
    .with_columns(
        (pl.col('revenue') / pl.col('region_total') * 100).alias('revenue_pct')
    )
    .filter(pl.col('revenue_pct') > 5)  # 超过区域 5% 的订单
    .select(['order_id', 'region', 'revenue', 'revenue_pct'])
    .collect()
)

# 计算移动平均（7天窗口）
df_with_ma = (
    df.sort('order_date')
    .with_columns(
        pl.col('daily_sales')
        .rolling_mean(window_size=7)
        .alias('sales_7d_ma')
    )
)

4.4 时间序列处理：动态分组与重采样

# 场景：每分钟 QPS 统计（2GB Nginx 日志秒级处理）
logs = pl.read_csv(
    'nginx_access.log',
    separator=' ',
    has_header=False,
    new_columns=['ip', 'timestamp', 'method', 'path', 'status', 'bytes'],
    schema={
        'timestamp': pl.Str,
        'status': pl.Int32,
        'bytes': pl.Int32,
    }
)

qps = (
    logs.lazy()
    .with_columns(
        pl.col('timestamp').str.to_datetime(format='%d/%b/%Y:%H:%M:%S')
    )
    .group_by_dynamic('timestamp', every='1m')
    .agg([
        pl.len().alias('requests'),
        pl.col('status').filter(pl.col('status') >= 500).len().alias('errors'),
        pl.col('bytes').sum().alias('bytes_total'),
    ])
    .with_columns(
        (pl.col('errors') / pl.col('requests') * 100).alias('error_rate')
    )
    .sort('timestamp')
    .collect()
)

print(qps)

4.5 多表 Join：CRM + 订单 + 物流三表融合

# 场景：构建客户全视图——CRM + 订单 + 物流
crm = pl.read_csv('crm.csv')
orders = pl.read_parquet('orders_2026.parquet')
logistics = pl.read_csv('logistics.csv')

# 自动识别 join key 类型
full_view = (
    orders.lazy()
    .join(crm, on='customer_id', how='left', suffix='_crm')
    .join(logistics, on='order_id', how='left', suffix='_logistics')
    .filter(pl.col('order_date') >= '2026-01-01')
    .with_columns(
        (pl.col('order_date').dt.offset_by('-1d') < pl.col('delivery_date'))
        .alias('is_delayed')
    )
    .group_by(['region', 'customer_tier', 'is_delayed'])
    .agg([
        pl.col('order_id').n_unique().alias('order_count'),
        pl.col('order_amount').sum().alias('gmv'),
        pl.col('delivery_days').mean().alias('avg_delivery_days'),
    ])
    .with_columns(
        (pl.col('gmv') / pl.col('order_count')).alias('avg_order_value')
    )
    .collect()
)

# 写出为 Parquet（压缩率比 CSV 高 5-10 倍）
full_view.write_parquet('customer_fullview.parquet', compression='zstd')

4.6 复杂条件逻辑：when-then-otherwise 链

# 场景：客户分层与标签体系
customers = (
    df.lazy()
    .group_by('customer_id')
    .agg([
        pl.col('order_amount').sum().alias('lifetime_value'),
        pl.len().alias('order_count'),
        pl.col('order_date').max().alias('last_order_date'),
    ])
    .with_columns(
        # 客户价值分层
        pl.when(pl.col('lifetime_value') >= 100000)
          .then(pl.lit('VIP'))
          .when(pl.col('lifetime_value') >= 10000)
          .then(pl.lit('Premium'))
          .when(pl.col('lifetime_value') >= 1000)
          .then(pl.lit('Regular'))
          .otherwise(pl.lit('Casual'))
          .alias('value_tier'),

        # 活跃状态（最近 30 天内有订单）
        (pl.col('last_order_date') >= pl.lit('2026-06-11'))
          .alias('is_active'),

        # 高频客户
        (pl.col('order_count') >= 20).alias('is_frequent'),
    )
    .with_columns(
        # 综合标签
        pl.concat_str([
            pl.col('value_tier'),
            pl.when(pl.col('is_active') & pl.col('is_frequent'))
              .then(pl.lit('-Champion'))
              .otherwise(pl.lit('')),
            pl.when(~pl.col('is_active') & (pl.col('lifetime_value') >= 10000))
              .then(pl.lit('-AtRisk'))
              .otherwise(pl.lit('')),
        ]).alias('customer_tag')
    )
    .collect()
)

4.7 字符串处理：正则与文本清洗

# 场景：日志数据清洗 + 结构化提取
logs = pl.read_csv('raw_logs.csv')

cleaned = (
    logs.lazy()
    .with_columns(
        # 提取 HTTP 方法
        pl.col('request').str.extract(r'^(GET|POST|PUT|DELETE|PATCH)\s+', 1).alias('method'),
        # 提取 URL path
        pl.col('request').str.extract(r'^\w+\s+(/\S*)', 1).alias('path'),
        # 提取 status code
        pl.col('request').str.extract(r'\s+(\d{3})\s+', 1).cast(pl.Int32).alias('status'),
        # 提取响应时间
        pl.col('request').str.extract(r'\s+(\d+)ms$', 1).cast(pl.Int32).alias('latency_ms'),
        # 清理 user agent
        pl.col('user_agent')
          .str.replace(r'Chrome/[\d.]+', 'Chrome')
          .str.replace(r'Firefox/[\d.]+', 'Firefox')
          .str.replace(r'Safari/[\d.]+', 'Safari')
          .str.replace(r'MSIE \d+', 'IE')
          .alias('browser'),
    )
    .filter(pl.col('status').is_not_null())
    .with_columns(
        # 路径规范化（去除查询参数）
        pl.col('path').str.split('?').list.get(0).alias('clean_path'),
        # 路径深度
        pl.col('clean_path').str.count_matches('/').alias('path_depth'),
    )
    .collect()
)

# 按 API endpoint 统计性能
endpoint_stats = (
    cleaned.lazy()
    .filter(pl.col('clean_path').str.starts_with('/api/'))
    .group_by('clean_path')
    .agg([
        pl.len().alias('request_count'),
        pl.col('latency_ms').mean().alias('avg_latency'),
        pl.col('latency_ms').max().alias('p99_latency'),
        pl.col('status').filter(pl.col('status') >= 400).len().alias('error_count'),
    ])
    .with_columns(
        (pl.col('error_count') / pl.col('request_count') * 100).alias('error_rate')
    )
    .sort('request_count', descending=True)
    .collect()
)

4.8 性能基准对比：Polars vs Pandas

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

# 生成测试数据（1000万行）
np.random.seed(42)
n = 10_000_000

pandas_df = pd.DataFrame({
    'customer_id': np.random.randint(1, 100_000, n),
    'product_id': np.random.randint(1, 10_000, n),
    'region': np.random.choice(['North', 'South', 'East', 'West'], n),
    'amount': np.random.exponential(100, n),
    'quantity': np.random.randint(1, 20, n),
    'order_date': pd.date_range('2024-01-01', periods=n, freq='min'),
})

# Polars 版本
polars_df = pl.from_pandas(pandas_df)

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

# 基准测试
p_time = benchmark('Pandas: GroupBy + sum', lambda: pandas_df.groupby('region')['amount'].sum())
r_time = benchmark('Polars: GroupBy + sum', lambda: polars_df.group_by('region').agg(pl.col('amount').sum()))
print(f'加速比: {p_time/r_time:.1f}x')

p_time = benchmark('Pandas: Multi-col + filter', lambda: pandas_df[(pandas_df['amount'] > 100)]['amount'].sum())
r_time = benchmark('Polars: Multi-col + filter', lambda: polars_df.filter(pl.col('amount') > 100).select(pl.col('amount')).sum())
print(f'加速比: {p_time/r_time:.1f}x')

p_time = benchmark('Pandas: Join (2x 50k)', lambda: pandas_df.merge(pandas_df[['customer_id', 'region']].drop_duplicates(), on='customer_id', how='left'))
r_time = benchmark('Polars: Join (2x 50k)', lambda: polars_df.join(polars_df.select(['customer_id', 'region']).unique(), on='customer_id', how='left'))
print(f'加速比: {p_time/r_time:.1f}x')

# 内存对比
import sys
p_mem = sys.getsizeof(pandas_df) / (1024**2)
r_mem = polars_df.estimated_size() / (1024**2)
print(f'\n内存占用: Pandas={p_mem:.1f}MB, Polars={r_mem:.1f}MB, 节省: {p_mem/r_mem:.1f}x')

典型基准测试结果（16 核 MacBook Pro M2，1000 万行数据）：

4.9 Pandas 到 Polars 迁移：渐进式策略

不建议你一次性把整个代码库迁移——风险太大。以下是推荐的渐进式迁移策略：

# 第一步：安装 + 导入共存
# pip install polars
import polars as pl
import pandas as pd

# 第二步：用 Polars 读取，用 to_pandas() 兼容老代码
def load_data(path):
    """新代码用 Polars，性能提升；但保持返回 pandas DataFrame 兼容老逻辑"""
    return pl.read_parquet(path).to_pandas()

# 第三步：逐步迁移热点函数
def compute_daily_metrics_polars(df: pl.DataFrame) -> pl.DataFrame:
    """把跑得最慢的那个函数单独迁移到 Polars"""
    return (
        df.lazy()
        .with_columns(
            (pl.col('revenue') / pl.col('quantity')).alias('unit_price'),
            pl.col('order_date').dt.date().alias('order_day'),
        )
        .group_by('order_day')
        .agg([
            pl.col('revenue').sum().alias('daily_gmv'),
            pl.col('order_id').n_unique().alias('daily_orders'),
            pl.col('customer_id').n_unique().alias('daily_customers'),
            pl.col('revenue').mean().alias('avg_order_value'),
        ])
        .with_columns(
            (pl.col('daily_gmv') / pl.col('daily_orders')).alias('avg_order_value_check')
        )
        .sort('order_day')
        .collect()
    )

# 第四步：完全迁移后的 DataFrame 管道
def etl_pipeline_polars():
    orders = pl.scan_parquet('s3://data-lake/orders/')
    products = pl.scan_parquet('s3://data-lake/products/')
    customers = pl.scan_parquet('s3://data-lake/customers/')

    return (
        orders
        .join(products, on='product_id', how='left', suffix='_product')
        .join(customers, on='customer_id', how='left', suffix='_customer')
        .filter(pl.col('status') == 'completed')
        .with_columns(
            pl.col('order_date').dt.date().alias('order_day'),
            (pl.col('price') * pl.col('quantity') * (1 - pl.col('discount'))).alias('net_revenue'),
        )
        .group_by(['order_day', 'category'])
        .agg([
            pl.col('net_revenue').sum().alias('gmv'),
            pl.len().alias('order_count'),
            pl.col('customer_id').n_unique().alias('unique_customers'),
        ])
        .sort(['order_day', 'gmv'], descending=[False, True])
        .sink_parquet('s3://data-warehouse/daily_metrics/')
    )

4.10 生产部署：流式 ETL 与云存储

# 场景：处理 S3 上的每日分区 Parquet 文件（分块，流式）
import polars as pl

# 读取 S3 Parquet（需要 s3fs: pip install s3fs）
daily_files = pl.scan_parquet(
    's3://data-lake/orders/year=2026/month=07/',
    glob=False,  # 禁用 glob 模式，显式指定分区
)

result = (
    daily_files
    .filter(pl.col('status') == 'completed')
    .group_by(['region', 'product_category'])
    .agg([
        pl.col('order_id').n_unique().alias('orders'),
        pl.col('net_amount').sum().alias('gmv'),
        pl.col('net_amount').mean().alias('aov'),
    ])
    .sort('gmv', descending=True)
    .sink_parquet('s3://data-warehouse/agg/orders_2026_07.parquet')
)

# 配置云存储：S3/GCS/Azure Blob
pl.Config.set_s3_access_key_id('your-key')
pl.Config.set_s3_secret_access_key('your-secret')
pl.Config.set_s3_endpoint('https://storage.googleapis.com')  # GCS 兼容

# 多文件并行读取
files = [f's3://bucket/partition={i}/*.parquet' for i in range(10)]
combined = pl.concat([pl.scan_parquet(f) for f in files])

五、性能优化：榨干 Polars 的每一分性能

5.1 懒加载优先

黄金法则：能用 scan_ 就不用 read_，能用 lazy() 就用 lazy()。

# 差：一次性全部加载到内存
df = pl.read_csv('data.csv')
result = df.filter(pl.col('x') > 0).group_by('cat').agg(pl.col('y').sum())

# 好：建立计划，Polars 自动优化后再执行
result = (
    pl.scan_csv('data.csv')    # 不实际读文件
    .filter(pl.col('x') > 0)   # 谓词下推：只在扫描时过滤
    .group_by('cat')
    .agg(pl.col('y').sum())
    .collect()                  # 实际执行
)

5.2 选择合适的文件格式

Parquet 格式优化参数：

# 写 Parquet 时选择压缩算法
df.write_parquet(
    'output.parquet',
    compression='zstd',      # Zstandard: 压缩率和解压速度俱佳
    # compression='snappy', # 压缩快，解压快，体积大
    # compression='gzip',   # 通用，兼容性最好
    row_group_size=100_000,  # 每个 Row Group 10 万行，平衡压缩和随机访问
)

5.3 数据类型优化

Polars 的数据类型选择直接影响性能和内存占用：

# 场景：给每个数值列选择合适的数据类型
df = (
    df
    .with_columns(
        # id 列用 UInt32，比 Int64 省一半内存
        pl.col('user_id').cast(pl.UInt32),
        # 精确金额用 Decimal，避免浮点精度问题
        pl.col('price').cast(pl.Decimal(precision=12, scale=2)),
        # 小整数值用 UInt8
        pl.col('quantity').cast(pl.UInt8),
        # 类别列用 Enum（枚举），比 String 省 5-10x 内存
        pl.col('status').cast(pl.Enum(['pending', 'active', 'completed', 'cancelled'])),
    )
)

# 日期时间优先用 Int64（毫秒时间戳）而非字符串
df = df.with_columns(
    pl.col('date_str').str.to_date().alias('date'),
    # 比 datetime 对象快，因为可以 SIMD 处理
)

# 验证内存节省
print(df.select(pl.all().sum_horizontal().alias('total_bytes')).to_series()[0])
# 原始 vs 优化后：一般能节省 50-70% 内存

5.4 大表 Join 优化

# 问题：大表 JOIN 大表，数据 shuffle 导致内存爆炸
# 方案：broadcast 小表，或先过滤再 join

# 广播小表（< 100MB 的表自动广播）
# Polars 自动识别小表，在每个分区内复制一份，不做 shuffle

# 优化：先过滤，减少 JOIN 的数据量
large_table = pl.scan_parquet('s3://data/large_table.parquet')
small_table = pl.read_parquet('metadata/small.parquet')

# 差：先 JOIN 再过滤
result = large_table.join(small_table, on='key').filter(pl.col('amount') > 1000)

# 好：先过滤，再 JOIN
result = (
    large_table
    .filter(pl.col('amount') > 1000)   # 减少 80% 数据量
    .join(small_table, on='key', how='left')
)

5.5 并行度调优

import polars as pl
import os

# 检测可用核心数
print(f"CPU 核心数: {os.cpu_count()}")
print(f"Polars 线程池: {pl.thread_pool_size()}")

# 对于 IO 密集型任务（读取云存储），可以增加并发
pl.Config.set_streaming_chunk_size(50_000)  # 默认 10 万行/批次

# 验证：查看实际执行计划
plan = (
    pl.scan_parquet('data.parquet')
    .filter(pl.col('x') > 0)
    .group_by('cat')
    .agg(pl.col('y').sum())
)
print(plan.explain(streaming=True))  # 打印流式执行计划

六、避坑指南：从 Pandas 迁移 Polars 必须知道的事

坑 1：没有行索引

Pandas 有 df.iloc[0] 和 df.loc['row_name']，Polars 没有。Polars 用 row 方法替代：

# Pandas
df.iloc[0]           # 第一行
df.iloc[-1]          # 最后一行
df.loc['row_name']   # 按索引名

# Polars：行号访问
df.row(0)            # 第一行（返回 tuple）
df.row(-1)           # 最后一行
df.row(0, named=True)  # 命名字典形式

# 如果需要筛选前 N 行，用 top_k
df.top_k(5, by='amount')  # amount 最大的 5 行

# 如果需要 nth 行
df.slice(0, 1)  # 取前 1 行

坑 2：apply 是性能杀手

# Pandas 思维：apply 一个 Python 函数
df['grade'] = df['score'].apply(lambda x: 'A' if x >= 90 else 'B')

# Polars 正确做法：使用原生表达式
df = df.with_columns(
    pl.when(pl.col('score') >= 90)
      .then(pl.lit('A'))
      .otherwise(pl.lit('B'))
      .alias('grade')
)

# 实在需要 Python 函数（性能降 10-50 倍）
df.with_columns(
    pl.col('score').map_elements(lambda x: complex_python_logic(x), return_dtype=pl.Float64)
)

坑 3：列名空格和大小写

# Polars 对列名大小写敏感
df.select(pl.col('Amount')) != df.select(pl.col('amount'))

# 列名有空格需要加反引号（字符串引号不行！）
df.select(pl.col('Order Amount'))  # 对
df.select(pl.col("Order Amount"))  # 错！这是字符串，不是列引用

# 最好像这样：保持列名规范（无空格、全小写）
df.columns = [c.lower().replace(' ', '_') for c in df.columns]

坑 4：lazy 模式下不能直接查看中间结果

# 调试 tip：先用 eager 模式跑通，再用 lazy 优化
# 差：直接在 lazy 模式下 print
print(df.lazy().filter(pl.col('x') > 0))  # 只打印计划，不打印数据

# 好：collect() 之后再检查
small_df = pl.read_csv('sample.csv')  # 抽一小部分数据
result = (small_df.lazy()
    .filter(pl.col('x') > 0)
    .group_by('cat')
    .agg(pl.col('y').sum())
    .collect())
print(result)  # 这时候才真的执行

# 然后把相同的逻辑应用到完整数据
final = (
    pl.scan_csv('full_data.csv')
    .filter(pl.col('x') > 0)
    .group_by('cat')
    .agg(pl.col('y').sum())
    .collect()
)

坑 5：空值处理语义不同

# Pandas: fillna 作用于整列
# Polars: 表达式链中有隐式空值传播

df = pl.DataFrame({'a': [1, None, 3], 'b': [4, 5, None]})

# Pandas
df['a'].fillna(0)  # OK
df['a'] + df['b']  # None + 5 = None（传播）

# Polars: fill_null 是显式方法
df.with_columns(
    pl.col('a').fill_null(0),
    (pl.col('a') + pl.col('b')).fill_null(0).alias('sum')  # 空值先传播，再填充
)

七、什么时候用 Polars，什么时候不用

应该用 Polars 的场景

继续用 Pandas 的场景

八、Polars 生态全景：Python 之外的版图

Polars 不只是一个 Python 库，它是一个完整的多语言数据框架：

Rust 原生：polars crate 直接用于高性能后端服务，数据管道无需 Python 解释器开销。

Node.js：通过 node-polars，Node.js 开发者可以享受同等的性能，用于构建 ETL 微服务或数据分析 API。

JavaScript/WASM：polars-wasm 让浏览器内数据处理成为可能——100MB 的 Parquet 文件可以在浏览器中直接分析，无需上传到服务器。

# Python 与 Rust 的互操作：PyO3 桥接
# Rust 代码中：
# #[pymodule]
# fn polars_rs(_py: Python, m: &PyModule) -> PyResult<()> {
#     m.add_class::<DataFrame>()?;
#     Ok(())
# }

# Python 中调用：速度与 Rust 原生几乎一致
import polars_rs  # 纯 Rust 实现，无 GIL

结语

Polars 不是一个「更好的 Pandas」——它是一个完全不同的物种。Pandas 是 Python 数据处理的起点，教会了整整一代人如何操作表格数据；但在 2026 年的数据规模下，它的设计假设已经不再成立。

Polars 的价值在于：它让你用几乎同样简洁的 API，获得接近 Rust 原生程序的数据处理性能。惰性求值把数据库查询优化的能力带到了 DataFrame 层面；Arrow 内存格式消除了 Python 数据科学中最浪费的序列化瓶颈；SIMD 向量化让 CPU 的每一赫兹主频都物尽其用。

如果你今天有一个跑了一夜的 Pandas 脚本，把它改成 Polars。你不需要重写架构，不需要换语言，不需要学新范式——只是把 df = pd.read_csv 改成 df = pl.scan_csv，把 df.groupby() 改成 df.lazy().group_by()，把 df.apply() 改成表达式链。

你的 ETL 脚本值得跑得更快。你值得把深夜的时间要回来。

本文代码在 Polars 1.42.1 下测试通过。所有基准数据基于 2026 年 7 月公开基准测试环境。