Polars 内存装不下时如何用 streaming 引擎执行查询并查看哪些算子落回内存【免费下载链接】polarsExtremely fast Query Engine for DataFrames, written in Rust项目地址: https://gitcode.com/GitHub_Trending/po/polars数据集大到超出可用内存时Polars 默认的collect()会把全部数据当作一个批次处理查询在内存峰值点上要求所有数据都装得下。解决办法是把 lazy 查询交给 streaming 引擎按批次执行再通过物理计划图确认哪些算子是以流式方式运行、哪些算子被回退fall back到内存引擎。适用前提查询以 lazy API 构建如pl.scan_csv返回的LazyFrame若要用show_graph出图需要在本机安装 Graphviz 并加入 PATH。确认默认执行方式为什么会把内存吃满先明确问题所在。Query execution 文档对默认collect的说明是With the defaultcollectmethod Polars processes all of your data as one batch. This means that all the data has to fit into your available memory at the point of peak memory usage in your query.也就是说如果查询中的某些步骤例如聚合前的中间结果需要超过可用内存默认执行就会失败或产生严重内存压力。文档对更大的数据集给出的路径是If your data requires more memory than you have available Polars may be able to process the data in batches usingstreamingmode. To use streaming mode you simply pass theenginestreamingargument tocollect注意这里文档用的是 may be able to——是否真的能按批次处理取决于查询里的算子是否都有流式实现这一点在后面的计划图中可以看到。用collect(enginestreaming)执行查询Streaming 文档给出的最小示例是import polars as pl q1 ( pl.scan_csv(docs/assets/data/iris.csv) .filter(pl.col(sepal_length) 5) .group_by(species) .agg(pl.col(sepal_width).mean()) ) df q1.collect(enginestreaming)把示例中的docs/assets/data/iris.csv替换成你自己的数据文件并在collect上传enginestreaming即可查询结构不需要其他改动。关于默认值有一个版本上的重要变化见 Version 2.0-rc 升级说明从 2.0 起lazy 查询的engineauto即不传参解析为 streaming 引擎而不再是内存引擎eager 的DataFrame操作不受影响仍解析到内存引擎。sink_*写文件本来就派发到 streaming 引擎不受该参数影响。如果某条查询你明确要强制回内存引擎执行文档给出的写法是 lf.collect(enginein-memory) # per query或者进程级设置 pl.Config.set_engine_affinity(in-memory) # process-wide以及环境变量POLARS_ENGINE_AFFINITYin-memory。这些写法同时适用于 SQL 路径。查看哪些算子落回内存引擎Streaming 文档明确说明Polars 可以把很多操作按流式方式运行但有些操作本质上不是流式的或暂时还没有流式实现。这种情况下 Polars 会对这些算子回退到内存引擎用户不需要感知这个回退但排查内存或性能问题时值得去看To inspect the physical plan of streaming query, you can plot the physical graph. The legend shows how memory intensive the operation can be.对应的操作是对查询调用show_graph指定plan_stagephysical和enginestreaming示例取自 streaming.pyq1 ( pl.scan_csv(docs/assets/data/iris.csv) .filter(pl.col(sepal_length) 5) .group_by(species) .agg( mean_widthpl.col(sepal_width).mean(), mean_width2pl.col(sepal_width).sum() / pl.col(sepal_length).count(), ) ) q1.show_graph(plan_stagephysical, enginestreaming)出图后按文档的描述来读图例legend标注了每个算子可能有多大的内存开销据此就能看出哪些节点是真正流式执行的、哪些会占用大块内存即回退到内存引擎执行的部分。如果不想弹窗查看也可以像文档示例那样把图落到文件q1.show_graph( plan_stagephysical, enginestreaming, showFalse, output_pathquery_plan.png, )output_path替换为你想保存图片的路径即可。还有一个版本相关的注意点2.0 中explain()和show_graph()默认不再渲染流式计划只有显式传入enginestreaming时才会输出 streaming 物理计划2.0 同时把show_graph的默认plan_stage从ir改为physical。所以核对算子时enginestreaming这个参数不能省。限制与结果核对行序不再保证streaming 引擎对不需要行序的算子unpivot、group_by、join 等不保证行顺序。如果你的下游代码依赖某种偶然顺序要么显式sort要么在算子支持时传maintain_order例如join(..., maintain_orderleft)。回退不是错误某个算子落回内存引擎意味着该算子执行时仍需要内存如果数据集对该算子确实装不下仅靠 streaming 不足以解决问题——此时应结合 Query execution 文档中的分片思路用scan子集 首尾.head/.collect在小数据上调试查询再回到全量数据执行。验证查询是否走通的方式就是collect(enginestreaming)正常返回 DataFrame算子级别的流式/回退情况以show_graph(plan_stagephysical, enginestreaming)出的物理计划图及其图例为准。相关文档入口Streaming、Query execution、Query plan、Version 2.0-rc 升级说明。【免费下载链接】polarsExtremely fast Query Engine for DataFrames, written in Rust项目地址: https://gitcode.com/GitHub_Trending/po/polars创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考