如何用Polars LazyFrame流式写入分区Parquet至云存储？

1. 用Polars的sink_parquet流式写入分区Parquet #

Polars的LazyFrame原生提供了sink_parquet方法，支持边处理边写入，无需将全量结果加载到内存。直接在处理后的LazyFrame上调用该方法，指定分区列、行组大小等参数，就能实现流式输出到云存储（如S3）。

示例代码：

import polars as pl

# 从S3读取数据构建LazyFrame
lf = pl.scan_pyarrow_dataset("s3://your-bucket/path/to/dataset")

# 执行数据处理逻辑
processed_lf = lf.filter(pl.col("value") > 100).select("id", "value", "category")

# 流式写入分区Parquet到S3
processed_lf.sink_parquet(
    "s3://your-output-bucket/parquet-output",
    partition_by=["category"],
    row_group_size=100_000,  # 控制行组大小，避免内存过载
    compression="snappy"
)

• sink_parquet会逐批处理数据，每完成一批就写入对应分区文件，全程不会将所有数据加载到内存。
• row_group_size可根据内存情况调整，平衡写入性能和内存占用。
• 需确保Polars具备云存储读写权限（通过环境变量配置AWS凭证、IAM角色等方式）。

2. 结合PyArrow Dataset API手动流式分区写入 #

如果需要更精细的自定义控制，可以借助PyArrow的Dataset API，配合Polars的流式迭代器实现：

import polars as pl
import pyarrow as pa
import pyarrow.dataset as pad

# 获取处理后的流式批次迭代器
streaming_iter = processed_lf.streaming().iter_batches(batch_size=100_000)

# 定义PyArrow分区规则和写入配置
partitioning = pad.partitioning(pa.schema([("category", pa.string())]))
write_options = pad.ParquetWriteOptions(row_group_size=100_000)

# 创建PyArrow Dataset写入器
dataset_writer = pad.dataset(
    "s3://your-output-bucket/parquet-output",
    format="parquet",
    partitioning=partitioning,
    write_options=write_options,
    existing_data_behavior="overwrite_or_ignore"
)

# 逐批写入数据
for batch in streaming_iter:
    pa_batch = batch.to_arrow()
    dataset_writer.write(pa_batch)

这种方式适合需要自定义分区逻辑、写入规则的场景，通过iter_batches逐批获取处理结果，再交给PyArrow写入分区路径，同样避免全量加载内存。

注意事项 #

• 确保Polars版本≥0.18：sink_parquet对云存储的完善支持从该版本开始，旧版本可能存在兼容性问题。
• 优化云存储写入效率：写入S3时，避免过小的批次导致过多API调用，可根据网络和内存情况调整batch_size或row_group_size。
• 权限配置：运行环境需具备S3读写权限，可通过环境变量设置AWS密钥，或在AWS服务（如EC2、EKS）中使用IAM角色。

内容的提问来源于stack exchange，提问作者shadowtalker