Polars 中高效逐行生成 DataFrame 并批量落盘的完整实践指南-Python教程

本文介绍在 polars 中处理流式逐行数据生成场景的最佳实践，重点对比列表累积、vstack 拼接等传统方式，推荐使用 lazyframe + `sink_csv` 的流式写入方案，并提供可直接复用的向量化批处理与自定义分解函数集成方法。

在实时数据采集、日志解析或模型推理等场景中，常需从生成器（generator）逐行获取原始数据，并经 decompose() 等逻辑提取结构化特征后，持续写入磁盘。此时若采用“每行建 DataFrame → vstack 拼接”或“Python 列表累积 → 定期转 DataFrame”，不仅内存开销大、性能差，还违背 Polars 的列式计算设计哲学。

✅ 最优解：LazyFrame 流式构建 + sink_csv 批量落盘
Polars 原生支持从可迭代对象（如生成器）直接构建 LazyFrame，并可通过 .sink_csv() 实现真正的流式、分块（batched）磁盘写入——无需将全部数据载入内存，且自动优化 I/O 与类型推断：

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 行刷盘一次（batch_size 控制内存峰值）
lf.sink_csv("output.csv", batch_size=100)

该方案优势显著：

• ✅ 内存恒定：仅缓存一个 batch（如 100 行）于内存；
• ✅ 零中间 DataFrame：避免 vstack 的重复内存分配与列对齐开销；
• ✅ 类型安全：schema 参数显式声明列类型，规避运行时推断错误；
• ✅ 可扩展：后续可无缝接入 .filter()、.with_columns() 等链式转换。

? 当 decompose() 逻辑复杂时：用 map_batches 向量化封装
若 decompose(row) 不是简单解包（如需调用外部 API、条件分支或状态依赖），可将其封装为 map_batches 的批处理函数。注意：务必设置 streamable=True 以启用流式执行：

def decompose(row):
    # 示例：对元组做非向量化变换（实际中应尽量向量化）
    a, b = row
    return a * 2, b ** 2

lf = (
    pl.LazyFrame({"raw": generation_mechanism()})
    .map_batches(
        lambda df: df.select(
            pl.col("raw").map_elements(
                decompose,
                return_dtype=pl.Struct({"feature_a": pl.Int64, "feature_b": pl.Int64})
            )
        ),
        streamable=True
    )
    .select(pl.col("raw").struct.unnest())  # 展开 struct 为独立列
)

lf.sink_csv("output.csv", batch_size=100)

⚠️ 不推荐的方案及原因