public void map(Object key, Text value, Context context) 
{
  ItemDataitemData = null; 
  List<AmazonCustomer>customerList = 
    AmazonCustomer.parseAItemLine(value.toString());
  if(customerList.size() == 0)
  {
    return;
  }
  for (AmazonCustomer customer : customerList) 
  {
    itemData = customer.itemsBrought.iterator().next();
    break;
  }


  String[] tokens = itemData.title.split("\s");
  for(String token: tokens)
  {
    if(token.length() > 3)
    {
      context.write(new Text(token), new IntWritable(itemData.salesrank));
    }
  }
}

public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException 
{
  doubleclickrate = 0; 
  for(IntWritableval: values)
  {
    if(val.get() > 1)
    {
      clickrate = clickrate + 1000/Math.log(val.get());
    }
    else
    {
      clickrate = clickrate + 1000;
    }
  }
  context.write(new Text("keyword:" +key.toString()), 
    newIntWritable((int)clickrate)); 
}

public void map(Object key, Text value, Context context) throws IOException, InterruptedException 
{
  String[] keyVal = value.toString().split("\s");
  if (keyVal[0].startsWith("keyword:")) 
  {
    context.write(
      new Text(keyVal[0].replace("keyword:", "")), 
      new Text(keyVal[1]));
  }
  else if (keyVal[0].startsWith("client")) 
  {
    List<String[]> bids = new ArrayList<String[]>();
    double budget = 0;
    String clientid = keyVal[0];
    String[] tokens = keyVal[1].split(",");
    for (String token : tokens) 
    {
      String[] kp = token.split("=");
      if (kp[0].equals("budget")) 
      {
        budget = Double.parseDouble(kp[1]);
      }
      else if (kp[0].equals("bid")) 
      {
        String[] bidData = kp[1].split("\|");
        bids.add(bidData);
      }
    }

    for (String[] bid : bids) 
    {
      String keyword = bid[0];
      String bidValue = bid[1];
      context.write(new Text(keyword), 
        new Text(new StringBuffer()
        .append(clientid).append(",")
      .append(budget).append(",")
      .append(bidValue).toString()));
    }
  }
}

public void reduce(Text key, Iterable<Text> values, 
  Context context) throws IOException, InterruptedException 
{
  String clientid = null;
  String budget = null;
  String bid = null;
  String clickRate = null;

  List<String>bids = new ArrayList<String>();
  for (Text val : values) 
  {
    if (val.toString().indexOf(",") > 0) 
    {
      bids.add(val.toString());
    }
    else
      {
        clickRate = val.toString();
      }
    }
    StringBufferbuf = new StringBuffer();
    for (String bidData : bids) 
    {
      String[] vals = bidData.split(",");
      clientid = vals[0];
      budget = vals[1];
      bid = vals[2];
      buf.append(clientid).append(",")
       .append(budget).append(",")
       .append(Double.valueOf(bid)).append(",")
       .append(Math.max(1, Double.valueOf(clickRate)));
      buf.append("|");
    }
    if (bids.size() > 0) 
    {
      context.write(key, new Text(buf.toString()));
    }
}