Составление операций с потоками в Scala

Question

Предположим, у вас есть программа, которая каким-то образом управляет потоком Stream[Foo], чтобы произвести вычисление, представляющее интерес, например.

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

Прекрасный, только теперь вы должны сделать некоторые другие виды вычислений на myFooStream как

myFooStream.map(toBar).sum 

А вы хотели бы, чтобы составить эти вычисления как-то так, что вам не нужно перебирать дважды над потоком (допустим, что по какой-то причине итерация по потоку дорогая).

Есть ли способ Scala-ish справиться с этой проблемой? Моя проблема, более абстрактно, заключается в том, что я хотел бы как-то абстрактно вычислить эти потоки из итерации по этим потокам. То есть лучше всего, если бы я мог как-то написать два метода f: Stream[Foo] => Bar и g: Stream[Foo] => Baz и как-то составить f и g таким образом, чтобы они работали на одной итерации потока.

Есть ли абстракция, которая позволяет это?

ОБНОВЛЕНО ВОПРОС: Я немного погубил. Будут ли подсказки scalaz полезными с этой проблемой?

Answer 1

поток не будет повторять дважды:

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 

Answer 2

Streams естественно стараются избегать создания их элементов несколько раз, если это возможно, memoizing результаты. Из docs:

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:

Нужен быть осторожным запоминание; вы можете очень быстро съесть большой объем памяти, если не будете осторожны. Причина этого в том, что memoization 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])(_ + _, _ :+ _)