本文介绍在 polars 中高效处理逐行数据流（如实时采集、流式解析）的最佳实践，重点对比列表累积、vstack 拼接等传统方式，推荐使用 lazyframe + `sink_csv` 流式落盘方案，并提供可直接运行的向量化批处理示例。
在实际数据工程场景中，常需从外部系统（传感器、API、日志流等）逐行获取原始数据，并经 decompose() 等函数解析为结构化特征，最终持久化为 CSV/Parquet 文件。此时，性能瓶颈往往不在于计算本身，而在于 DataFrame 构建与内存管理策略。下面我们将从效率、内存友好性与代码可维护性三方面，系统梳理最优解法。
❌ 不推荐：手动维护 Python 列表（低效但常见）
feature_a_list = []
feature_b_list = []
for row in generation_mechanism():
a, b = decompose(row)
feature_a_list.append(a)
feature_b_list.append(b)
# ... 最终构建 DataFrame
df = pl.DataFrame({"feature_a": feature_a_list, "feature_b": feature_b_list})问题：Python list.append() 在 CPython 中虽快，但大量小对象频繁分配会引发 GC 压力；且无法利用 Polars 的零拷贝内存布局优势。当 decompose() 返回非标量（如嵌套结构），还需额外序列化开销。
❌ 更不推荐：逐行 vstack（严重性能陷阱）
df = pl.DataFrame(schema={"feature_a": pl.String, "feature_b": pl.String})
for row in generation_mechanism():
new_df = pl.DataFrame({"feature_a": [a], "feature_b": [b]})
df = df.vstack(new_df) # O(n) 每次复制整个 DataFrame！关键缺陷：vstack 是线性时间复杂度操作——第 k 次调用需复制前 k−1 行数据。累计 10 万行时，总拷贝量达 ~50 亿行等效量，内存与 CPU 开销爆炸式增长。Polars 官方文档明确警告：“Never use vstack in loops”。
✅ 推荐方案一：LazyFrame + sink_csv（流式、零中间内存）
这是 Polars 专为流式场景设计的原生高性能方案，无需显式缓冲，自动分批写入：
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"]
)
# 流式写入磁盘，batch_size 控制每批次行数（默认 1024）
lf.sink_csv("output.csv", batch_size=100)✅ 优势：
- 内存恒定：仅缓存当前 batch（如 100 行），无全量 DataFrame 占用；
- 零 Python 层循环：全部在 Rust 运行时执行，避免 GIL 争用；
- 支持并行：sink_csv 底层自动启用多线程压缩与 I/O；
- 可扩展：后续可无缝接入 sink_parquet、sink_ndjson 或数据库 sink。
⚠️ 注意：sink_csv 要求 Polars ≥ 0.20.0，且生成器需返回同构元组/字典（每行字段名与类型一致）。
✅ 推荐方案二：批处理 + map_elements（灵活适配复杂 decompose）
若 decompose() 逻辑较重（如正则提取、JSON 解析、调用外部服务），可结合 itertools.batched 实现向量化批处理：
from itertools import batched
import polars as pl
def decompose(row): # 示例：简单转换，实际可含复杂逻辑
return row[0] * 2, row[1] ** 2
flush_threshold = 100
for batch in batched(generation_mechanism(), flush_threshold):
# 批量转 Series → 向量化 map → 结构化解包
df = (
pl.Series("raw", list(batch))
.map_elements(decompose, return_dtype=pl.List(pl.Int64))
.list.to_struct(fields=["feature_a", "feature_b"])
.struct.unnest()
)
# 此时 df 为标准 DataFrame，可 write_csv / write_parquet
df.write_csv("output.csv", append=True) # append=True 需确保文件已存在✅ 优势：
- map_elements 在批内并行执行（启用 n_threads 参数可进一步加速）；
- list.to_struct + unnest 避免 Python 循环，保持 Polars 内存连续性；
- 显式控制 batch 大小，平衡内存与 I/O 效率。
? 关键注意事项
- 避免 map_elements 中的全局状态：因 Polars 可能重排或并行执行批次，decompose() 必须是纯函数；
- 类型声明至关重要：为 map_elements 显式指定 return_dtype（如 pl.Struct([pl.Field("a", pl.Int64)])），否则推断失败将降级为慢速 Python 模式；
- CSV 追加写入：write_csv(append=True) 仅支持追加，且首行需手动处理列头（建议首次写入时 header=True，后续 header=False）；
- 替代存储格式：对大数据集，优先选用 sink_parquet（列式压缩、Schema 自动推导、查询更快）。
总结
| 方案 | 内存占用 | 速度 | 适用场景 |
|---|---|---|---|
| 手动列表累积 | 中 | 中 | 小规模（ |
| vstack 循环 | 极高 | 极慢 | ❌ 绝对避免 |
| LazyFrame.sink_csv | 最低 | 最快 | 标准流式场景（推荐首选） |
| 批处理 map_elements | 低 | 快 | decompose 逻辑复杂、需定制化处理 |
终极建议：优先采用 LazyFrame 流式方案；若需在 decompose 中集成非向量化逻辑（如调用外部 API），再选用批处理模式，并始终通过 pl.Config.set_streaming_chunk_size() 调优批大小。
