Polars 中高效逐行生成并批量写入 DataFrame 的最佳实践-Python教程

本文介绍在 polars 中处理流式逐行数据生成场景的最优方案，重点推荐基于 lazyframe 的 `sink_csv` 流式写入、`batched` 批量构造及 `map_elements` 向量化预处理，避免低效的逐行 vstack 或重复 list 追加。

在实际数据工程中，常遇到外部系统（如传感器、API 流、日志解析器）以逐行方式持续产出原始记录，需经 decompose() 等逻辑提取结构化特征，并高效累积为 Polars DataFrame 后定期落盘（如 CSV）。此时，传统“每行建 DataFrame + vstack”或“纯 Python list 累积再转 DataFrame”的做法存在明显性能瓶颈：前者因频繁内存分配与元数据重建导致 O(n²) 时间复杂度；后者虽内存友好，但缺乏 Polars 原生向量化优势，且手动管理 flush 逻辑易出错。

✅ 推荐方案一：LazyFrame + sink_csv（首选，真正流式）
Polars 的 LazyFrame 支持从可迭代对象（包括生成器）直接构建，并通过 sink_csv() 实现零拷贝、分批、内存可控的流式写入，无需显式维护中间 DataFrame：

import polars as pl

def generation_mechanism():
    for row in external_data_stream():  # 如 requests.iter_lines()、kafka consumer 等
        yield row

# 直接从生成器构建 LazyFrame（不触发计算）
lf = pl.LazyFrame(generation_mechanism(), schema=["raw_row"])

# 使用 map_batches + vectorized decompose（关键优化）
def decompose_batch(df: pl.DataFrame) -> pl.DataFrame:
    # 假设 decompose 可向量化：输入 Series，输出 struct 列
    return df.select(
        pl.col("raw_row")
        .map_elements(lambda x: (x["id"] * 2, x["value"].upper()), 
                      return_dtype=pl.Struct({"feature_a": pl.Int64, "feature_b": pl.String}))
        .struct.unnest()
    )

lf = lf.map_batches(decompose_batch, streamable=True)
lf.sink_csv("output.csv", batch_size=10_000)  # 自动按 10k 行分块写入

⚠️ 注意：map_batches 中的函数需标记 streamable=True 且避免非流式操作（如 sort, join），确保流式执行；decompose 应尽可能向量化（用 pl.col().str.xxx / pl.col().dt.xxx 替代 map_elements）。

✅ 推荐方案二：itertools.batched + 批量构造（兼容性强，适合复杂 decompose）
若 decompose() 逻辑难以向量化，可借助 Python 标准库 batched 分组，再对每批数据统一处理，显著减少 DataFrame 构造次数：

from itertools import batched
import polars as pl

flush_threshold = 500

for batch in batched(generation_mechanism(), flush_threshold):
    # 批量应用 decompose（仍为逐行，但仅调用 N/batch_size 次）
    processed = [decompose(row) for row in batch]

    # 一次性构造 DataFrame（高效！）
    df = pl.DataFrame(
        processed,
        schema={"feature_a": pl.Int64, "feature_b": pl.String}
    )

    # 追加写入（注意：CSV 不支持原生追加，需用 'a' 模式并确保无 header）
    with open("output.csv", "a") as f:
        if f.tell() == 0:  # 首次写入添加 header
            df.write_csv(f, include_header=True)
        else:
            df.write_csv(f, include_header=False)

❌ 应避免的方案