本文介绍如何在数据逐行生成场景下，以最高效率将结果写入 polars dataframe 并批量持久化到磁盘，重点推荐基于 `lazyframe` + `sink_csv` 的流式处理方案，并对比传统列表累积与逐行 vstack 的性能缺陷。
在实时数据采集、日志解析或流式 ETL 等场景中，常需从生成器（如 generation_mechanism()）逐行获取原始数据，并通过 decompose(row) 提取结构化特征，最终写入磁盘（如 CSV）。此时若采用传统方式——如累积 Python 列表后构造 DataFrame 或 每行新建 DataFrame 并 vstack 拼接——会显著降低性能：前者受 Python 列表动态扩容与内存拷贝拖累；后者因频繁创建小 DataFrame 及 vstack 的 O(n) 合并开销，在百万级数据下极易成为瓶颈。
✅ 最优解：使用 pl.LazyFrame + sink_csv 流式落盘
Polars 原生支持惰性计算与流式写入。只需将生成器直接传入 pl.LazyFrame，再调用 sink_csv 并指定 batch_size，即可实现零中间内存堆积、按批自动刷盘：
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)该方案优势明显：
- ✅ 零显式内存缓冲：无需维护 feature_a_list 等临时列表；
- ✅ 无 DataFrame 构造开销：避免 pl.DataFrame(...) 频繁调用；
- ✅ 自动批处理与磁盘 I/O 优化：batch_size 控制写入粒度，兼顾吞吐与延迟；
- ✅ 天然支持流式计算链：可无缝接入 map_batches、filter、select 等惰性操作。
? 当 decompose() 逻辑较复杂时：结合 map_batches 向量化处理
若 decompose() 不是简单解包，而是含业务逻辑（如字符串解析、数值转换），应优先将其向量化（例如改用 map_elements + streamable=True），再嵌入 LazyFrame 流程：
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) # 显式声明返回类型提升性能
).alias("features")
),
streamable=True # 关键！启用流式执行
)
.select(
pl.col("features").list.to_struct(fields=["feature_a", "feature_b"])
)
.unnest("features")
)
lf.sink_csv("output.csv", batch_size=100)⚠️ 注意事项与避坑指南
- ❌ 避免 vstack 循环：data = data.vstack(new_row_df) 时间复杂度为 O(n)，n 行需 O(n²) 总耗时；
- ❌ 谨慎使用 map_elements：若未设 return_dtype 或未启用 streamable=True，可能触发全量 materialization；
- ✅ 优先向量化 decompose：尽量用 Polars 原生表达式（如 str.split(), dt.strftime()）替代 Python 函数；
- ✅ 小批量测试：首次使用 sink_csv 时建议先用 batch_size=10 验证输出格式与性能；
- ? 替代格式：如需更高吞吐，可换用 sink_parquet()（列式存储，压缩率高，读写更快）。
综上，LazyFrame + sink_csv 是 Polars 生态中处理逐行生成数据的官方推荐范式——它将“生成→转换→落盘”抽象为声明式流水线，既保障性能，又保持代码简洁与可维护性。
