Polars 与 DuckDB：单机现代数据处理栈

引言

很长一段时间，“大数据"意味着必须上集群。但现实是：大多数团队的日常分析任务，数据量在几 GB 到几百 GB 之间，跑在一台配置不错的机器上，用对工具比堆机器更划算。Polars 与 DuckDB 正是这一趋势的代表——它们让单机处理能力提升了整整一个数量级。

不是所有数据都需要分布式；单机 + 列式 + 向量化 + 多核并行，能覆盖绝大多数分析场景。

本文讲清两件事：Polars 与 DuckDB 各自强在哪，以及如何把它们组合成一套高效的单机数据栈。

一、单机处理的复兴

1.1 为什么单机又香了

硬件在变强（几十核 CPU、几百 GB 内存、NVMe SSD），而分布式系统有固定开销（调度、网络、序列化、协调）。当数据量没有大到跨越单机能力时，分布式的复杂度纯属负担。

1.2 三个关键技术的成熟

• 列式内存布局：按列存储，聚合与扫描只读需要的列，缓存友好。
• 向量化执行：一次处理一批（batch）而非一行，充分利用 SIMD 与 CPU 流水线。
• 多核并行：现代引擎默认吃满所有核心，无需手动分区。

1.3 Arrow 作为共同底座

Polars 与 DuckDB 都基于 Apache Arrow 内存格式，因此数据可以在两者之间零拷贝传递。这也是它们能无缝协作的根本原因。

# Arrow 的价值
# - 列式、零拷贝、跨语言
# - Polars/DuckDB/pandas(2.0+) 共享内存格式
# - Parquet/Arrow 文件与内存布局同构, IO 高效

二、Polars：列式 DataFrame 与查询优化

2.1 表达式 API

Polars 的核心是表达式（Expression）：你把计算描述成表达式，引擎负责优化与并行。这与 pandas 的"立即执行"模型完全不同。

import polars as pl

df = pl.read_parquet("orders.parquet")
result = (
    df.lazy()
      .filter(pl.col("amount") > 0)
      .group_by("user_id")
      .agg([
          pl.col("amount").sum().alias("total"),
          pl.len().alias("cnt"),
      ])
      .sort("total", descending=True)
      .head(10)
      .collect()
)

2.2 惰性求值

.lazy() 返回一个查询计划，只有在 .collect() 时才执行。中间引擎可以做谓词下推、投影下推、公共子表达式消除等优化，甚至把过滤条件推到 Parquet 读取层。

plan = (
    pl.scan_parquet("s3://lake/orders/*.parquet")
      .filter(pl.col("dt") == "2026-10-01")     # 下推到扫描
      .select(["user_id", "amount"])            # 只读两列
)
print(plan.explain())     # 查看优化后的执行计划
plan.sink_parquet("out.parquet")  # 流式写出, 不占内存

explain() 能看到优化器做了什么，是排查性能问题的第一工具。

2.3 内存与性能特征

Polars 用 Arrow 列式内存，字符串列也做了高效编码。相比 pandas 的 object dtype，字符串处理快得多、内存占用也低。

三、DuckDB：嵌入式分析数据库

3.1 定位

DuckDB 是进程内（in-process）的 OLAP 数据库，可以理解为"分析版的 SQLite”。它没有独立服务进程，作为一个库嵌入你的应用，但提供完整的 SQL 与事务。

import duckdb

con = duckdb.connect("analytics.duckdb")
con.sql("""
    CREATE TABLE orders AS
    SELECT * FROM read_parquet('lake/orders/*.parquet')
""")
con.sql("""
    SELECT user_id, sum(amount) AS total
      FROM orders
     WHERE dt = DATE '2026-10-01'
     GROUP BY user_id
     ORDER BY total DESC
     LIMIT 10
""").show()

3.2 直接查询文件

DuckDB 最实用的特性之一是直接查询 Parquet/CSV/JSON，无需先导入。它把文件当作外部表，利用谓词下推与列裁剪只读必要数据。

-- 直接聚合 S3 上的 Parquet, 无需建表
SELECT date_trunc('day', ts) AS d, count(*) AS n
  FROM read_parquet('s3://lake/events/*.parquet')
 WHERE ts >= TIMESTAMP '2026-10-01'
 GROUP BY 1 ORDER BY 1;

3.3 扩展与生态

DuckDB 通过扩展支持 Parquet、JSON、Iceberg、Delta、HTTP/S3 等。它还能作为计算引擎被其他工具调用，例如在 Python、R、Java 中嵌入。

# 常用扩展
# INSTALL iceberg; LOAD iceberg;   -- 直接读湖表
# INSTALL httpfs;  LOAD httpfs;    -- 访问 S3/HTTP
# INSTALL json;    LOAD json;      -- JSON 解析
# ATTACH 'lake.duckdb' AS other;   -- 多库挂载

四、两者协作：分工与互操作

4.1 分工模型

# Polars: 程序内的数据变换、特征工程、复杂列计算
# DuckDB: SQL 分析、多表 JOIN、即席查询、文件直查
# 两者共享 Arrow, 可零拷贝互转

4.2 零拷贝互操作

import polars as pl
import duckdb

# Polars → DuckDB: 注册为视图, 零拷贝
df = pl.read_parquet("orders.parquet")
con = duckdb.connect()
con.register("orders", df)          # Arrow 零拷贝
con.sql("SELECT user_id, sum(amount) FROM orders GROUP BY 1").pl()

# DuckDB → Polars: 直接转
res = con.sql("SELECT * FROM orders LIMIT 100").pl()

这种互操作让"用 Polars 做变换、用 DuckDB 做 SQL"成为自然的组合，而不是二选一。

4.3 组合工作流

# 典型工作流: DuckDB 抽数 → Polars 变换 → DuckDB 落地
import polars as pl, duckdb

con = duckdb.connect()
raw = con.sql("""
    SELECT * FROM read_parquet('lake/raw/events/*.parquet')
    WHERE dt BETWEEN DATE '2026-09-01' AND DATE '2026-10-01'
""").pl()                                   # 抽数

features = (
    raw.lazy()
       .with_columns(
           pl.col("ts").dt.hour().alias("hour"),
           (pl.col("amount") * pl.col("qty")).alias("gmv"),
       )
       .group_by(["user_id", "hour"])
       .agg(pl.col("gmv").sum())
       .collect()                           # 变换
)

con.register("features", features)
con.sql("COPY features TO 'lake/features.parquet' (FORMAT PARQUET)")

五、性能原理：向量化、惰性求值与并行

5.1 向量化执行

传统逐行处理每行都有函数调用开销；向量化一次处理一批（如 2048 行），把循环展开、用上 SIMD，CPU 利用率大幅提升。

# 逐行 vs 向量化
# 逐行: for row in rows: acc += f(row)   → 每行一次调用
# 向量化: acc = sum(batch_array)          → 一次处理一批
# DuckDB/Polars 都是向量化引擎

5.2 惰性求值与计划优化

惰性让引擎在真正执行前看到整个计划，从而做全局优化：先过滤再 JOIN、只读需要的列、把常量折叠。立即执行的 pandas 做不到这一点。

5.3 多核并行

两者默认使用所有 CPU 核心。Polars 的 group_by 会按 key 分区并行聚合；DuckDB 的算子按 morsel 分片并行。这意味着单机性能几乎随核心数线性增长，直到 IO 成为瓶颈。

六、实战：从 pandas 迁移到 Polars

6.1 心智模型转变

从 pandas 迁移到 Polars，首先要转变心智模型：

• 布尔筛选：pandas 的 df[df.a > 0] 对应 Polars 的 df.filter(pl.col("a") > 0)
• 分组聚合：pandas 的 df.groupby("k").agg(...) 对应 Polars 的 df.group_by("k").agg(...)
• 新增列：pandas 的 df.assign(b=...) 对应 Polars 的 df.with_columns(...)
• 链式赋值改为表达式组合，立即执行改为惰性求值加 collect()
• 列引用从字符串或属性改为 pl.col("name") 表达式

6.2 常见迁移片段

# pandas
# df["gmv"] = df["amount"] * df["qty"]
# res = df.groupby("user").agg(gmv=("gmv", "sum")).reset_index()

# polars
res = (
    df.lazy()
      .with_columns((pl.col("amount") * pl.col("qty")).alias("gmv"))
      .group_by("user")
      .agg(pl.col("gmv").sum())
      .collect()
)

6.3 迁移注意事项

• 索引消失：Polars 无行索引，需显式用列表达顺序，如 with_row_index()。
• 空值语义：Polars 区分 null 与 NaN，聚合时注意 drop_nulls()。
• 类型严格：不会隐式转换字符串与数值，需显式 cast()。
• 字符串操作：用 .str. 命名空间，性能远优于 pandas。

七、典型场景与反模式

7.1 适合的场景

# [x] 单机 ETL: 读 Parquet → 变换 → 写 Parquet
# [x] 特征工程: 大规模列计算, 多核并行
# [x] 即席分析: DuckDB 直查文件, 无需建仓
# [x] 数据探查: 快速 profile 大文件
# [x] 本地测试: 用真实数据子集验证逻辑
# [x] 数据导出: 生成下游需要的聚合表

7.2 反模式

• 用 Polars 做分布式：它不是分布式引擎，超内存要退化为流式，而非加机器。
• DuckDB 当在线事务库：它是 OLAP，不适合高并发点查与写入。
• 反复 scan 同一文件：每次都重读 IO，应物化中间结果。
• pandas 写法硬套 Polars：逐行 apply 会退化为慢路径，应改用表达式。
• 忽略内存上限：collect() 全量物化，超大结果应 sink_parquet 流式写出。

八、选型与边界

8.1 何时用哪个

8.2 单机的边界

单机的上限由内存与磁盘决定。经验边界：内存 2-5 倍以内的数据可全量物化；更大时应依赖流式执行（Polars sink_*、DuckDB 的 out-of-core）或分区处理。当数据量持续超过单机数倍、或需要多用户并发查询时，才考虑转向分布式引擎。

8.3 与湖仓的关系

Polars 与 DuckDB 不是湖仓的替代品，而是湖仓的本地加速器：湖仓存 PB 级数据，单机栈处理 GB 到 TB 级子集与本地迭代，两者通过 Parquet/Iceberg 无缝衔接。

总结

Polars 与 DuckDB 代表了数据处理的"单机复兴"：列式内存、向量化执行、惰性优化、多核并行，让一台机器能做的事远超过去。它们不是要取代分布式系统，而是让绝大多数分析任务不必动用分布式。真正用好它们的关键，是理解惰性求值与向量化的原理，避免把 pandas 习惯和分布式思维硬套进来。

参考与延伸阅读

• Polars 官方用户指南：表达式、惰性求值与流式 API
• DuckDB 官方文档：SQL 方言、扩展与文件直查
• Apache Arrow 官方文档：列式内存格式与互操作
• 数据查询引擎与向量化执行 — 向量化原理深入
• 数据湖技术 — 单机栈与湖仓的衔接
• 数据 SQL 查询优化 — 计划优化方法论
• ETL 与 ELT 设计 — 单机 ETL 的定位