SalesReducer.java

package com.godoro.hadoop.example;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;


public class SalesReducer extends Reducer<Text,SalesWritable, Text,SalesWritable>{

  private SalesWritable aggregateWritable = new SalesWritable();
  @Override
  protected void reduce(Text key, Iterable<SalesWritable> values, Context context)
          throws IOException, InterruptedException {
    int valueCount=0;
    int salesQuantitySum=0;
    double salesPriceSum=0;
    double lineAmountSum=0;
    for(SalesWritable value : values) {
      salesQuantitySum+=value.getSalesQuantity();
      salesPriceSum+=value.getSalesPrice();
      lineAmountSum+=value.getLineAmount();
      valueCount++;
    }
    double salesPriceAverage=salesPriceSum/valueCount;
    aggregateWritable.setSalesQuantity(salesQuantitySum);
    aggregateWritable.setSalesPrice(salesPriceAverage);
    aggregateWritable.setLineAmount(lineAmountSum);
    context.write(key, aggregateWritable);
  }
  
}