在 Scala 中对流进行组合操作答案

【问题标题】：Composing Operations on Streams in Scala在 Scala 中对流进行组合操作
【发布时间】：2015-04-19 02:32:32
【问题描述】：

假设您有一个程序以某种方式操纵流 Stream[Foo] 以产生感兴趣的计算，例如

myFooStream.map(toBar).groupBy(identity).mapValues(_.size)

很好，除了现在你必须在 myFooStream 上做一些其他类型的计算，比如

myFooStream.map(toBar).sum

并且您希望以某种方式组合这些计算，这样您就不需要对流进行两次迭代（假设由于某种原因，对流进行迭代很昂贵）。

有一些 Scala 式的方法来处理这个问题吗？更抽象地说，我的问题是我想以某种方式从对这些流的迭代中抽象出对这些流的计算。也就是说，如果我能以某种方式编写两个方法 f: Stream[Foo] => Bar 和 g: Stream[Foo] => Baz 并以某种方式组合 f 和 g 以使它们在流的单次迭代上运行，那么最好的办法是。

是否有一些抽象允许这样做？

更新的问题：我已经做了一些挖掘。 scalaz 箭头对这个问题有帮助吗？

【问题讨论】：

标签： scala stream

【解决方案1】：

Streams 通过记忆结果自然会尽量避免多次生成元素。来自docs：

Stream 类还使用了记忆功能，以便将先前计算的值从 Stream 元素转换为 A 类型的具体值。

我们可以看到，通过构造一个Stream，它会在每次生成元素时打印，并运行多个操作：

val stream = Stream.from(0).map(x => { println(x); x }).take(10) //prints 0
val double = stream.map(_ * 2).take(5).toList //prints 1 through 4
val sum = stream.sum //prints 5 through 9
val sum2 = stream.sum //doesn't print any more

只要您使用val 而不是def，这将有效：

只要有东西抓住头，头就会抓住尾巴，所以它会递归地继续。另一方面，如果头部没有任何东西（例如，我们使用def 来定义Stream），那么一旦不再直接使用它，它就会消失。

这个备忘录意味着你必须小心Streams：

一个人必须小心记忆；如果你不小心，你会很快吃掉大量的内存。原因是Stream 的记忆创建了一个很像scala.collection.immutable.List 的结构。

当然，如果项目的生成不是什么昂贵的，但Stream的实际遍历，或者因为太昂贵而无法使用记忆，可以随时使用foldLeft一个元组，跟踪多个值：

//Only prints 0-9 once, even if stream is a def
val (sum, double) = stream.foldLeft(0 -> List.empty[Int]) { 
    case ((sum, list), next) => (sum + next, list :+ (next * 2)) 
}

如果这是一个足够常见的操作，您甚至可以丰富 Stream 以使一些更常见的操作（例如 foldLeft、reduceLeft 和其他一些以这种格式可用：

implicit class RichStream[T](val stream: Stream[T]) extends AnyVal {
    def doubleFoldLeft[A, B](start1: A, start2: B)(f: (A, T) => A, g: (B, T) => B) = stream.foldLeft(start1 -> start2) { 
        case ((aAcc, bAcc), next) => (f(aAcc, next), g(bAcc, next)) 
    }
}

这将允许您执行以下操作：

val (sum, double) = stream.doubleFoldLeft(0, List.empty[Int])(_ + _, _ :+ _)

【讨论】：

谢谢。我没有意识到溪流一直在头上。 doubleFoldLeft 很整洁，但在我的具体情况下，它会变得复杂。我需要将大量的计算链接在一起。如果没有强大的抽象，这将变得笨拙。
另外，当我尝试使用 foldLeft 跟踪多个值时，我意识到：我失去了高阶函数的好处。换句话说，虽然我通常可以编写类似 myStream.groupBy(identity).mapValues(_.size) 的内容来创建计数映射，但编写带有元组的 foldLeft 意味着我必须负责增加映射。所以我下降到一个较低的抽象层次，我也想吃我的功能蛋糕！ :)
你肯定会在这里丢失一些高阶函数，但在要求极高效率或非常具体的优化时这是很自然的。您确定需要这种级别的优化吗？ Stream's 自然记忆就够了吗？您是否考虑过使用较低级别的构造，例如 for 理解？

【解决方案2】：

流不会迭代两次：

Stream.continually{println("bob"); 1}.take(4).map(v => v).sum
bob
bob
bob
bob
4

和

val bobs = Stream.continually{println("bob"); 1}.take(4)
val alices = Stream.continually{println("alice"); 2}.take(4)
bobs.zip(alices).map{ case (b, a) => a + b}.sum
bob
bob
bob
bob
alice
alice
alice
alice
12

【讨论】：