CODEWITHSUNDEEP
hadoopmapreducehashmap
HHH
HHH

Reputation: 6465

Merging two SortedMapWritable in Hadoop?

I have defined a class called EquivalenceClsAggValue which has a data field of array (called aggValues). 

class public class EquivalenceClsAggValue extends Configured implements WritableComparable<EquivalenceClsAggValue>{

public ArrayList<SortedMapWritable> aggValues;

It has a method which take another object of type EquivalenceClsAggValue and merge its aggValues into aggValues of this class as follows:

public void addEquivalenceCls(EquivalenceClsAggValue eq){
    //comment: eq contains only one entry as it comes from the mapper

    if (this.aggValues.size()==0){ //new line
        this.aggValues = eq.aggValues;  
                    return;     
    }

    for(int i=0;i<eq.aggValues.size();i++){

        SortedMapWritable cm = aggValues.get(i); //cm: current map
        SortedMapWritable nm = eq.aggValues.get(i); //nm: new map
        Text nk = (Text) nm.firstKey();//nk: new key

        if(cm.containsKey(nk)){//increment the value
            IntWritable ovTmp = (IntWritable) cm.get(nk);
            int ov = ovTmp.get();
            cm.remove(nk);
            cm.put(nk, new IntWritable(ov+1));
        }
        else{//add new entry
            cm.put(nk, new IntWritable(1));
        }


    }
}

But this function is not merging two aggValues. Could someone help me figure it out? This is how I call this method:

public void reduce(IntWritable keyin,Iterator<EquivalenceClsAggValue> valuein,OutputCollector<IntWritable, EquivalenceClsAggValue> output,Reporter arg3) throws IOException {

        EquivalenceClsAggValue comOutput = valuein.next();//initialize the output with the first input

        while(valuein.hasNext()){
            EquivalenceClsAggValue e = valuein.next();
            comOutput.addEquivalenceCls(e);             
        }           
        output.collect(keyin, comOutput);
    }

Upvotes: 0

Views: 365

Answers (1)

Chris White
Chris White

Reputation: 30089

Looks like you're falling foul of object re-use. Hadoop re-uses the same object so each call to valuein.next() actually returns the same object reference, but the contents of that object are re-initialised via the readFields method.

Try changing as follows (create a new instance to aggregate into):

 EquivalenceClsAggValue comOutput = new EquivalenceClsAggValue();

 while(valuein.hasNext()){
   EquivalenceClsAggValue e = valuein.next();
   comOutput.addEquivalenceCls(e);             
 }           
 output.collect(keyin, comOutput);

EDIT: and you probably need to update your aggregate method too (to be wary of object re-use):

public void addEquivalenceCls(EquivalenceClsAggValue eq){
  //comment: eq contains only one entry as it comes from the mapper

  for(int i=0;i<eq.aggValues.size();i++){

    SortedMapWritable cm = aggValues.get(i); //cm: current map
    SortedMapWritable nm = eq.aggValues.get(i); //nm: new map
    Text nk = (Text) nm.firstKey();//nk: new key

    if(cm.containsKey(nk)){//increment the value
        // you don't need to remove and re-add, just update the IntWritable
        IntWritable ovTmp = (IntWritable) cm.get(nk);
        ovTmp.set(ovTmp.get() + 1);
    }
    else{//add new entry
        // be sure to create a copy of nk when you add in to the map
        cm.put(new Text(nk), new IntWritable(1));
    }
  }
}

Upvotes: 1

Related Questions