Python 是否有像 Java 8 Stream/Kotlin Sequence 这样的标准流畅迭代器运算符？

Question

我是一名 Java/Kotlin 开发人员，明年将用 Python 速成课程来教授课程。我有点震惊，因为我没有找到任何函数结构来处理链式运算符的迭代。这可以在 Java 中通过 Streams 找到，如下所示：

Stream<MyItem> filtered = myItems.stream().filter(item -> item.isSelected());

或者带有序列的 Kotlin：

val filtered = myItems.asSequence().filter { it.isSelected }

我认为 Scala 也有类似的东西。我知道可以在 Python 中实现流畅的接口，并且RxPy通过反应式编程带来这些迭代运算符。但是，是否没有固有的语言功能、标准实用程序或通用库以基于拉动的方式完成这些运算符，例如

map()

、

filter()

、

flatMap()

等？

Answer 1

有一个库可以准确实现您正在寻找的内容：

pyxtension 位于 https://github.com/asuiu/pyxtension，它也已在 PyPi 上做好生产准备和维护。 通常的功能性 Py 代码如下所示：

lst = xrange(1,6)
reduce(lambda x, y: x * y, map(lambda _: _ * _, filter(lambda _: _ % 2 == 0, lst)))

可以重写为：

the_stream = stream( xrange(1,6) )
the_stream.\
    filter(lambda _: _ % 2 == 0).\
    map(lambda _: _ * _).\
    reduce(lambda x, y: x * y)

它支持多线程映射，称为

fastmap

，以及多进程映射，称为

mpmap

。 它已准备好并开箱即用，大量使用 Python 迭代器的惰性求值，因此您不会像 Pandas DataFrame 那样将所有内容加载到内存中进行处理。这个 pyxtension.streams 对于处理 CSV、Lined Json、MessagePack、DataBase 游标等数据流要方便得多...

有了它，您将轻松使用所有 CPU 核心进行并行处理，作为 GIL 限制的解决方法。 只需运行

pip install pyxtension

即可安装。

Answer 2

Stream

模块的

streamable

类完成这项工作。在流上应用操作是lazy，并返回一个child流而不修改父流。

pip install streamable

integers: Stream[int] = Stream(lambda: range(10))

odd_square_strings: Stream[str] = (
    integers
    .map(lambda x: x ** 2)
    .filter(lambda x: x % 2 == 1)
    .map(str)
)

list(odd_square_strings)

['1', '9', '25', '49', '81']

支持：

• 并发
• 分批/展平
• 速率限制
• 异常处理
• 记录迭代进度