[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

State for moving average calculation

Hi all,

I’m currently calculating a moving average with DataStreams via:

            .keyBy(new XXXKeySelector())
            .aggregate(new MovingAverageAggregator(10))

MovingAverageAggregator uses a MovingAverageAccumulator to do all of the real work.

But if I want to use AggregatingState to support checkpointing, it seems like I’d have to create a new RichFunction (e.g. MovingAverageFunction) that implements Checkpointing, creates the AggregatingState in its initializeState() method, and then the stream creation is:

            .keyBy(new XXXKeySelector())
            .map(new MovingAverageFunction(10))

Yes? Seems kind of…awkward to have a function with state that is created using a descriptor that takes an aggregator which has an accumulator.

Also, is the state in this case the serialized accumulator object?


— Ken

Ken Krugler
+1 530-210-6378
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra