SalesMapper.java

package com.godoro.hadoop.example;

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SalesMapper extends Mapper<Object, Text, Text, SalesWritable> {

  private Text product = new Text();
  private SalesWritable salesWritable = new SalesWritable();

  
  public void map(Object key, Text value, Context context)
          throws IOException, InterruptedException {
    
    StringTokenizer lines=new StringTokenizer(value.toString(),"\r\n");
    while(lines.hasMoreTokens()){
      String line=lines.nextToken();
      String[] cells=line.split(",");
      product.set(getString(cells[0])+ " "+getString(cells[1]) );
      salesWritable.setSalesQuantity(getInt(cells[2]));
      salesWritable.setSalesPrice(getDouble(cells[3]));
      salesWritable.setLineAmount(getDouble(cells[4]));
      context.write(product, salesWritable);
    }
  }
  private static String getString(String string){
    int startIndex=0;
    if(string.startsWith("\"")){
      startIndex=1;
    }
    int endIndex=string.length();
    if(string.endsWith("\"")){
      endIndex=string.length()-1;
    }
    return string.substring(startIndex,endIndex);
  }
  private static double getDouble(String string){
   String stripped=getString(string);
   return Double.parseDouble(stripped);
  }
  private static int getInt(String string){
   String stripped=getString(string);
   return Integer.parseInt(stripped);
  }
}