0

我已经定义了一个名为的类EquivalenceClsAggValue，它具有数组的数据字段（称为aggValues）。

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

public ArrayList<SortedMapWritable> aggValues;

它有一个方法，该方法采用另一个类型的对象EquivalenceClsAggValue并将其合并aggValues到aggValues此类中，如下所示：

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));
        }


    }
}

但是这个函数并没有合并两个aggValues。有人可以帮我弄清楚吗？这就是我调用此方法的方式：

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);
    }
4

1 回答 1

1

看起来你正在犯规对象重用。Hadoop 重用相同的对象，因此每次调用valuein.next()实际上返回相同的对象引用，但该对象的内容通过 readFields 方法重新初始化。

尝试如下更改（创建一个新实例以聚合到）：

 EquivalenceClsAggValue comOutput = new EquivalenceClsAggValue();

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

编辑：你可能也需要更新你的聚合方法（要警惕对象重用）：

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));
    }
  }
}
于 2013-01-13T21:43:27.157 回答