implement mapReduce aggregator