Tried with reduce and it's giving me pretty weird results that make no sense
ie: 1 number for an entire RDD
val smaStream= inputStream.reduce{ case(t1,t2) =>
{
val sma= average.addDataPoint(t1)
sma
}}
Tried with transform and that worked correctly, but unfortunately, it works 1
RDD at a time so the moving average is reset when the next consecutive RDD is
read .. as if a new instance of the Average class is created for each RDD.
Is there a way to have 1 global variable of custom type (ie my case Average
type) .. somewhat like accumulators, but not incrementable in parallel - it
wouldn't make sense for a moving average.
The reason I want to apply a moving average function to a stream is so that
the tuples remain in Spark and benefit from its fault tolerant mechanisms.
My guess is that currently this is not possible, but I'll wait for one of the
Spark creators to comment on this.
-A
From: Benjamin Black [mailto:[email protected]]
Sent: March-26-14 11:50 AM
To: [email protected]
Subject: Re: closures & moving averages (state)
Perhaps you want reduce rather than map?
On Wednesday, March 26, 2014, Adrian Mocanu
<[email protected]<mailto:[email protected]>> wrote:
I'm passing a moving average function during the map phase like this:
val average= new Sma(window=3)
stream.map(x=> average.addNumber(x))
where
class Sma extends Serializable { .. }
I also tried to put creation of object average in an object like I saw in
another post:
object Average {
val smaFn = new VSRTXYSimpleMovingAverage[(String, Long)](3)
}
Every time average.addNumber is called it is a new instance.
How can I preserve state of average object?
Thanks
-Adrian
