本文介绍三种优化 Polars 并行读取 S3 上多个 Parquet 文件的方法：合并后统一过滤、异步并发采集，以及 collect_all 的适用边界，兼顾性能、内存与 schema 一致性。
本文介绍三种优化 polars 并行读取 s3 上多个 parquet 文件的方法：合并后统一过滤、异步并发采集，以及 `collect_all` 的适用边界，兼顾性能、内存与 schema 一致性。
在使用 Polars 处理分布式存储（如 Amazon S3）上的大量 Parquet 文件时，简单循环调用 pl.scan_parquet(url).filter(...) 并通过 pl.collect_all() 汇总，虽语义清晰，但默认行为往往未充分利用 I/O 和 CPU 并行性——尤其当文件数量多、单文件较小、网络延迟显著时，串行或弱并行策略会成为瓶颈。
以下是三种经过实践验证的高效替代方案，按推荐优先级排序：
✅ 方案一：合并扫描 + 统一过滤（推荐用于同构数据）
若所有 Parquet 文件结构一致（相同列名、类型），且最终目标是聚合结果（而非保留各文件独立 DataFrame），应优先采用 pl.concat() 合并扫描对象再统一计算：
import polars as pl
# 所有 URL 对应的 lazy frame 合并为一个逻辑视图
combined_scan = pl.concat(
[pl.scan_parquet(url) for url in urls],
how="vertical", # 要求 schema 完全一致
rechunk=False # 避免不必要的内存拷贝
)
# 单次过滤 + 单次 collect —— 减少计划构建开销与多次调度
result_df = combined_scan.filter(expr).collect()优势：
- Polars 查询优化器可对整个逻辑计划做全局优化（如谓词下推、列裁剪）；
- 避免重复解析元数据与打开连接；
- 实际 I/O 可由底层 Arrow/IO 层自动并行（取决于配置与文件分块）；
- 内存更友好（流式处理，无需缓存多个中间 lazy frame）。
⚠️ 注意：此方式要求所有 Parquet 文件 schema 严格一致。若存在字段缺失或类型冲突，
pl.concat()将抛出异常。可通过pl.scan_parquet(url).schema预检或启用schema_overrides显式对齐。
✅ 方案二：异步并发采集（推荐用于异构或需独立结果）
当需要保留每个文件的独立结果（如分片统计、校验）、或 schema 不完全一致时，asyncio.gather + collect_async() 提供真正的并发执行：
import asyncio
import polars as pl
async def fetch_and_filter(url: str):
return await pl.scan_parquet(url).filter(expr).collect_async()
# 并发启动所有任务（非阻塞）
results = await asyncio.gather(*[fetch_and_filter(url) for url in urls])
# results 是 list[pl.DataFrame]，顺序与 urls 一致优势：
- 网络 I/O 高度并发（尤其在 S3 高延迟场景下收益明显）；
- 每个
collect_async()在独立线程中执行，避免 GIL 限制； - 返回原始 DataFrame 列表，语义与
collect_all一致但更快。
⚠️ 注意：需确保事件循环已正确配置（如在 Jupyter 中使用
await或asyncio.run()）；同时注意并发数过高可能触发 S3 请求限频，建议结合aioboto3或polars内置连接池参数（如pl.Config.set_streaming_chunk_size()）进行调优。
⚠️ 方案三：collect_all 的合理使用场景
pl.collect_all() 本身支持一定程度的并行（依赖 Polars 后端线程池），但在以下情况仍适用：
- 文件数量较少（ 100MB）；
- 各文件需差异化处理（不同
expr或后续分支逻辑）； - 已启用 Polars 的 streaming 模式（
.collect(streaming=True)）且内存充足。
此时可微调以提升效率：
# 显式启用多线程（默认已开启，但可确认） pl.Config.set_verbose(True) # 查看执行计划是否并行化 pl.collect_all(queries, parallel=True) # Polars ≥ 0.20.16 默认 True
总结建议
| 场景 | 推荐方案 | 关键考量 |
|---|---|---|
| 同构数据 + 聚合结果 | pl.concat() + .filter().collect() |
最优查询优化、最低开销 |
| 异构数据 + 独立结果 |
asyncio.gather() + collect_async()
|
真实并发、灵活控制 |
| 小批量 + 简单逻辑 |
collect_all() + parallel=True
|
开箱即用，无需重构 |
无论选择哪种方式，均建议配合 Polars 的 profiling 工具验证实际性能：
pl.Config.set_verbose(True) pl.Config.set_streaming_chunk_size(10_000) # 流式处理小文件更高效
最终性能取决于数据分布、网络带宽、S3 配置及表达式复杂度——务必在真实环境中基准测试（%timeit 或 polars.show_versions() 辅助诊断）。
