安装 · 快速开始 · 示例 · 引擎 · 文档 文档 · 浏览器演练场 · 更新日志 · 贡献指南 用惰性流水线处理 Python 对象和异步输入，支持有界并发工作和显式资源清理。 版本 2.4.1 重整了 planner 和执行的内部实现，公开 API 没有变化。 版本 2.4.0 增加了有状态流水线操作、组合聚合、共享执行 限制、文件恢复，以及带版本信息的执行元数据。其范围见 更新日志， 批处理和 executor 示例见 性能指南。 路线图 列出了接下来的优化候选及其验收标准。 版本 2.5.0 在 Python 3.14 和 3.15 上，保留了按 key 归约、Pair 收集，以及内存中 Python join 和分组时的自定义 key 哈希错误。 2.5.0 的改动包括一个静态平均类型修正， 并给出了明确的迁移路径。这个版本新增了 带类型的异步 fold、 显式平均操作， 以及针对空值过滤的可选严格查找策略。 这些 API 不在 2.4.1 中。对类型检查器来说，Flow.reduce 仍然返回 Any， 即使回调带有类型标注也是如此。迁移示例见 reduce 指南， 验证相关的贡献者检查也见该指南。 python -m pip install fpstreams 预编译 wheel 包含原生扩展。从源码构建需要 Rust 1.85 或更新版本。 async extra 会安装 aiofiles，用于异步文件适配器。核心异步 流水线不需要 extra。其他适配器按需安装： python -m pip install "fpstreams[async]" # 异步文件适配器所需的 aiofiles python -m pip install "fpstreams[arrow]" # PyArrow 和 Parquet python -m pip install "fpstreams[data]" # NumPy、pandas 和 PyArrow python -m pip install "fpstreams[polars]" # Polars 和 PyArrow 快速开始 把一个可迭代对象传给 flow()，然后消费结果： from fpstreams import flow result = flow([1, 2, 3]).filter(lambda value: value > 1).map(lambda value: value * 10).to_list() assert result == [20, 30] 异步回调使用异步流水线， 分组使用 record 流水线。 引擎指南有助于诊断执行路径； 处理较大输入时会应用资源限制。 下面这个流水线取三个匹配的值，并在返回前关闭其生成器。 生成器会在 finally 块运行时打印，因此清理过程是可见的： from fpstreams import flow def numbers(): try: yield from range(10) finally: print("source closed") result = flow(numbers()).filter(lambda value: value % 2 == 0).take(3).to_list() print(result)

source closed

[0, 2, 4]

分组聚合见 record 流水线， 并发调用见 异步流水线。 如果只是简短的本地转换，用推导式通常就够了。fpstreams 适用于流水线需要分组聚合、提前终止、 资源清理，或带并发限制的异步工作的场景。小型流水线可能 会稍慢一些，因为规划和调度会带来额外开销。 惰性执行并不意味着每个操作都使用常量内存。排序、 分组和 join 需要额外的状态；当输入可能无法完全放入内存时， 请使用文档中说明的限制和溢写选项。 fpstreams 2 取代了 v1 的实现，并保留下面列出的兼容 别名。 Flow[T]：惰性值和 record 流水线的主要同步入口， 包括保留的表格数据源。 AsyncFlow[T]：带并发上限、排序、 超时、合并、去抖，以及清理流水线所创建任务的异步转换。 Rows[T]：用于表达式、join、分组、重塑和 I/O 的 record 视图。 Pairs[K, V]：键/值转换，以及按 key 的收集或聚合。 Collector 和 Aggregator：单遍归约，包括命名的 多重聚合。 Option 和 Result：带类型的值和错误容器。 自动执行规划：融合的 Python 循环、Arrow 原生前缀、 对受支持的标量计划使用原生 Rust 内核，对受保护的关系型 子路径也使用原生内核；当只有部分计划为原生时，采用混合执行。 为外部排序和分区 join/分组提供可配置的内存和输出限制。 版本 2.1 增加了执行报告、从 flow() 直接进行 record 操作、 标准 Arrow/dataframe 协议路由、保留的 NumPy 执行，以及新的 异步队列、预取、窗口和数值终端 API。发布摘要见 更新日志。 CI 矩阵面向标准的 CPython 3.11 到 3.15。 自由线程 CPython 3.14t 和 3.15t 包含在一个 实验性的、非阻塞的 CI 矩阵中，该矩阵会构建原生扩展并检查 核心并发契约。这些构建不是发布 wheel 的目标； 可选适配器的兼容性不在该矩阵范围内。 流水线是惰性的。转换操作构建计划；to_list()、aggregate()、first() 和 count() 这类终端操作负责执行计划。 from fpstreams import flow, item squares = ( flow(range(1, 10)) .filter(item % 2 == 0) # 保留偶数。 .map(item * item) # 对每个值求平方。 .take(3) # 取到三个结果后停止。 .to_list() ) assert squares == [4, 16, 36] 上面的占位符表达式等价于两个 lambda，但当整个计划都受支持时， 它也可以由原生引擎编译。 from fpstreams import agg, flow summary = flow([1, 2, 3, 4]).aggregate( count=agg.count(), total=agg.sum(), mean=agg.mean(), ) assert summary == {"count": 4, "total": 10, "mean": 2.5} 所有命名聚合共享对源数据的一次遍历。 flow() 是值和 record 的主要同步入口。与 Flow 语义不冲突的 record 操作，例如 with_columns() 和 group_by()，会自动进入 Rows 视图。 Record 可以是字典、dataclass、命名元组，或带属性的对象。 from fpstreams import agg, col, flow orders = [ {"region": "eu