本文介绍在 polars 中高效处理逐行生成数据的三种主流方案，重点推荐基于 lazyframe 的 `sink_csv` 流式写入方法，并对比列表累积、vstack 拼接等传统方式的性能差异与适用边界。
在实时数据采集、日志解析或流式 ETL 场景中，常需从逐行生成器（如传感器读数、API 分页响应、自定义迭代器）持续提取结构化特征并写入磁盘。若直接使用 pl.DataFrame() 逐行构造再 vstack 合并，将引发严重性能退化——因 Polars 的 DataFrame 是不可变的列式结构，每次 vstack 都需复制全部已有数据，时间复杂度为 O(n²)，完全不可扩展。
✅ 推荐方案：LazyFrame + sink_csv（流式写入）
Polars 原生支持零拷贝流式数据管道，核心是 pl.LazyFrame 结合 sink_csv() 方法：
import polars as pl
def generation_mechanism():
for x in range(1_000_000):
yield (x, x + 1)
# 直接从生成器构建 LazyFrame（不触发计算）
lf = pl.LazyFrame(generation_mechanism(), schema=["feature_a", "feature_b"])
# 流式写入 CSV，每 100 行刷新一次缓冲区（无需手动管理内存）
lf.sink_csv("output.csv", batch_size=100)✅ 优势：
Knowledge Graph - Csv Graph Loader Generator
下载
从 CSV 数据集生成图数据库加载器和三元组映射。将表格 CSV 数据转换为图数据库可直接使用的节点、边和三元组。
- 内存恒定：不加载全量数据到内存，仅按 batch_size 分块处理；
- 零手动管理：无需维护临时列表、计数器或文件打开/追加逻辑；
- 原生优化：底层调用 Rust 的流式 CSV writer，比 Python 文件 I/O 快 3–5×；
- 可扩展：支持 sink_parquet、sink_ndjson 等多种格式，且兼容 collect().write_database()。
⚙️ 处理复杂 decompose() 逻辑
若 decompose(row) 不是简单解包，而是含业务逻辑（如正则提取、嵌套字典解析），可借助 map_batches + map_elements（注意启用 streamable=True）：
def decompose(row):
# 示例：对元组做非向量化变换
return row[0] * 2, row[1] ** 2
lf = (
pl.LazyFrame({"raw": generation_mechanism()})
.map_batches(
lambda df: df.select(
pl.col("raw").map_elements(decompose, return_dtype=pl.List(pl.Int64))
),
streamable=True # 关键！启用流式执行
)
.select(
pl.col("raw")
.list.to_struct(fields=["feature_a", "feature_b"])
)
.unnest("raw")
)
lf.sink_csv("output.csv", batch_size=100)⚠️ 注意：map_elements 仍是逐行执行，但因其运行在 Polars 的流式调度器中，配合 batch_size 可有效摊销开销。如需极致性能，应重写 decompose 为向量化操作（例如用 str.extract 替代 re.search）。
❌ 不推荐方案对比
| 方案 | 时间复杂度 | 内存占用 | 问题 |
|---|---|---|---|
| 列表累积 + 批量构建 DataFrame | O(n) | O(n) | ✅ 简单可靠，适合中小规模（ |
| 逐行 vstack | O(n²) | O(n²) | ❌ 严重反模式，10k 行即明显卡顿，应绝对避免 |
| itertools.batched + Series.map_elements | O(n) | O(batch_size) | ⚠️ 折中方案，适合需精细控制批次逻辑的场景，但代码冗长且不如 sink_csv 原生 |
? 最佳实践总结
- 首选 LazyFrame.sink_csv(..., batch_size=N)：语义清晰、性能最优、错误率最低；
- 避免任何 DataFrame.vstack() 循环：这是 Polars 新手最常见性能陷阱；
- 慎用 map_elements：仅当无法向量化时使用，并始终设置 return_dtype 提升类型推断效率；
- 小批量测试先行：用 lf.head(1000).collect() 验证逻辑正确性，再切换至 sink_*；
- 磁盘 I/O 优化：确保目标路径为 SSD 或高性能存储，CSV 写入瓶颈常在磁盘而非 CPU。
通过拥抱 Polars 的惰性求值与流式 sink 机制，你不仅能写出更简洁、健壮的代码，更能轻松应对百万级甚至十亿级行数据的实时写入需求。
