Processing Streaming Data - Real-Time Analytics - page 179

retrieving the results of an aggregation that is using local memory as its
backing store, such as the MemoryMapState class.
At the moment, the MemoryMapState class is the only State
implementation built into Trident. It is primarily intended for testing
purposes with other State objects implemented to persist to more durable
stores, such as memcached .
Partition Local Aggregation
Although the normal aggregation operations imply a global partitioning
event, the partitionAggregate method allows for partitioning:
public class KeyDoubleAggregator
extends BaseAggregator<Map<Object, Double>> {
private static final long serialVersionUID = 1L;
public Map<Object, Double> init(Object batchId,
TridentCollector collector) {
return new HashMap<Object,Double>();
}
public void aggregate(Map<Object, Double> val,
TridentTuple tuple,
TridentCollector collector) {
Object key = tuple.get(0);
if (val.containsKey(key))
val.put(key, val.get(key) + tuple.getDouble(1));
else
val.put(key, tuple.getDouble(1));
}
public void complete(Map<Object, Double> val,
TridentCollector collector) {
for (Entry<Object,Double> e : val.entrySet()) {
collector.emit( new
Values(e.getKey(),e.getValue()));
}
}
}