高效构建并流式写入 Polars DataFrame 的最佳实践-Python教程

本文介绍如何在数据逐行生成场景下，以最高效率将结果写入 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)

⚠️ 注意事项与避坑指南