Polars中scan_pyarrow_dataset()不稳定原因及分区CSV扫描替代方案咨询

一、scan_pyarrow_dataset()标注「不稳定」的原因 #

• 接口兼容性风险：该函数依赖PyArrow Dataset的API，而PyArrow自身的Dataset模块仍在快速迭代，Polars为了适配PyArrow的后续更新，保留了调整函数参数、返回逻辑的空间，不承诺向后兼容。
• 整合逻辑未完全成熟：Polars与PyArrow Dataset的查询优化整合（比如分区过滤下推、列投影下推）仍在打磨中，部分边缘场景（如混合文件格式Dataset、复杂嵌套分区）的处理可能存在潜在问题，标注不稳定是提醒用户后续版本可能有重构。
• 功能优先级：Polars团队当前的核心优先级是完善原生的CSV分区扫描能力，scan_pyarrow_dataset()作为过渡性的兼容方案，暂时未纳入稳定API范畴。

二、无需全量加载内存的替代方案 #

方案1：PyArrow Scanner + Polars LazyFrame #

利用PyArrow Dataset的scanner()创建懒加载的扫描器，再传入Polars的scan_pyarrow_dataset()，既保留PyArrow的分区自动识别能力，又能利用Polars的Lazy API做查询优化，且不会全量加载数据：

import pyarrow.dataset as ds
import polars as pl

# 初始化PyArrow Dataset
dataset = ds.dataset("your/partitioned_csv_dir", format="csv")

# 创建扫描器，支持过滤、列投影下推（按需添加）
scanner = dataset.scanner(
    filter=ds.field("partition_col") == "target_value",
    columns=["col1", "col2", "partition_col"]
)

# 转为Polars LazyFrame，执行懒加载查询
lazy_df = pl.scan_pyarrow_dataset(scanner)
result = lazy_df.group_by("partition_col").agg(pl.sum("col1")).collect()

方案2：Polars原生scan_csv + 手动分区处理 #

如果可以手动识别分区规则（如Hive风格的key=value目录结构），可以用Polars的scan_csv结合路径解析来实现分区扫描，完全不依赖PyArrow的不稳定对接：

import polars as pl
import glob

# 递归获取所有CSV文件路径
csv_files = glob.glob("your/root_dir/**/*.csv", recursive=True)

# 从路径提取分区列（示例为year=2023/month=12的Hive格式）
def extract_partitions(file_path):
    path_parts = file_path.split("/")
    partitions = {}
    for part in path_parts:
        if "=" in part:
            key, val = part.split("=", 1)
            partitions[key] = val
    return partitions

# 为每个文件创建带分区列的LazyFrame
lazy_frames = []
for file in csv_files:
    part_cols = extract_partitions(file)
    lf = pl.scan_csv(file).with_columns(**part_cols)
    lazy_frames.append(lf)

# 合并所有LazyFrame并执行查询
combined_lf = pl.concat(lazy_frames, how="vertical_relaxed")
result = combined_lf.filter(pl.col("year") == "2023").collect()

方案3：PyArrow Dataset分批读取 #

如果需要逐批处理数据（如ETL场景），可以用PyArrow Dataset的to_batches()方法分批加载，每批转成Polars DataFrame处理，严格控制内存占用：

import pyarrow.dataset as ds
import polars as pl

dataset = ds.dataset("your/partitioned_csv_dir", format="csv")

# 按批次读取，batch_size可根据内存调整
for batch in dataset.to_batches(batch_size=10000):
    df = pl.from_arrow(batch)
    # 执行单批次处理逻辑
    processed_df = df.filter(pl.col("value") > 100)
    # 输出或存储处理结果
    processed_df.write_parquet(f"output/batch_{batch.index}.parquet")

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