closures & moving averages (state)

classic Classic list List threaded Threaded
3 messages Options
Reply | Threaded
Open this post in threaded view
|

closures & moving averages (state)

amoc

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

 

Reply | Threaded
Open this post in threaded view
|

Re: closures & moving averages (state)

Benjamin Black
Perhaps you want reduce rather than map?

On Wednesday, March 26, 2014, Adrian Mocanu <[hidden email]> 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

 

Reply | Threaded
Open this post in threaded view
|

RE: closures & moving averages (state)

amoc

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:[hidden email]]
Sent: March-26-14 11:50 AM
To: [hidden email]
Subject: Re: closures & moving averages (state)

 

Perhaps you want reduce rather than map?

On Wednesday, March 26, 2014, Adrian Mocanu <[hidden email]> 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