State for moving average calculation

[Ken Krugler]

Hi all,

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

            .keyBy(new XXXKeySelector())
            .window(GlobalWindows.create())
            .trigger(CountTrigger.of(1))
            .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?

Thanks,

— Ken

--------------------------
Ken Krugler
[phone]
http://www.scaleunlimited.com
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra