Aggregate Scala Stream multiple times with overlapping segments

Question

How can I make the following function tail recursive and fully streaming?

def aggregate[A, B](in: Stream[A], predicate: Stream[A] => B): Stream[B] = {
    in match {
        case Empty => Stream.empty
        case head #:: Empty => predicate(in) #:: Stream.empty
        case head #:: tail => predicate(in) #:: aggregate(tail, predicate)
    }
}

The goal process the stream with some lookahead. As an example let's say we have a stream of sequential numbers: val s: Stream[Int] = Stream.from(0). Now I'd like to calculate the sum of 5 sequential elements of this stream beginning at every element. With the function above it would look like this: val r: Stream[Int] = aggregate(s, _.take(5).sum), producing the stream of 10, 15, 20.... But this will break the stack on large input. How can I optimize this function?

score 2 · Accepted Answer · answered Mar 18 '15 at 03:07

2

Why do you say this will break the stack on larger input? Have you tried this? Please refer to this SO question, where Ken Bloom points out that streams don't need tail recursion.

In fact, I ran this on a take 1,000,000 and it returned in < 1 second.

answered Mar 18 '15 at 03:07

Justin Pihony

66,056
18
147
180

Right you are right, I have to confess that I didn't try this code on large input because I was 100% sure that it will break the stack. – ak. Mar 18 '15 at 03:23

Aggregate Scala Stream multiple times with overlapping segments

1 Answers1