1

我正在尝试使用 SequenceFile 在两个 mapReduce 程序之间传递数据。我要传递的数据格式为 >. 由于某种原因，地图中的某些条目似乎没有从一个程序传递到另一个程序。这是我的代码，首先是生成 de SequenceFileOutput 的 reducer，然后是从中读取的映射器。

公共静态类 IntSumReducer 扩展 Reducer {

public void reduce(Text key, Iterable<Text> values, 
                   Context context
                   ) throws IOException, InterruptedException {

    MapWritable vector = new MapWritable() ;

    for (Text val : values){
        if(vector.containsKey(val)){
            vector.put(val , new IntWritable(((IntWritable)vector.get(val)).get() + 1));
        }
        else
            vector.put(val , new IntWritable(1));
    }

    context.write(key, vector);

        }
    }

和映射器：

公共静态类 TokenizerMapper 扩展 Mapper{

  private final static int cota = 100;
  private final static double ady = 0.25;

  public void map(Text key, MapWritable value, Context context
          ) throws IOException, InterruptedException {

      IntWritable tot = (IntWritable)value.get(key);

      int total = tot.get();


      if(total > cota){
          MapWritable vector = new MapWritable() ;
          Set<Writable> keys = value.keySet();

          Iterator<Writable> iterator = keys.iterator();
          while(iterator.hasNext()){
              Text llave = (Text) iterator.next();
              if(!llave.equals(key)){
                  IntWritable cant = (IntWritable) value.get(llave);
                  double rel = (((double)cant.get())/(double)total);
                  if(cant.get() > cota && rel > ady ){
                      vector.put(llave, new DoubleWritable(rel));
                  }
              }
          }
          context.write(key,vector);     
      }
  }

}

4

1 回答 1

1 
for (Text val : values){
    if(vector.containsKey(val)){
        vector.put(val , new IntWritable(((IntWritable)vector.get(val)).get() + 1));
    }
    else
        vector.put(val , new IntWritable(1));
}

这就是你的问题 - val Text 对象被 hadoop 重用，所以在调用 vector.put 时，你应该创建一个新的 Text 对象以脱离 val 引用（其值将在 for 循环的下一次迭代中改变）。

您可以将您的逻辑修改为以下内容，它应该可以工作（我还更新了计数器增量逻辑以提高效率）：

IntWritable tmpInt;
for (Text val : values){
    tmpInt = (IntWritable) vector.get(val);

    if(tmpInt == null) {
        tmpInt = new IntWritable(0);
        // create a copy of val Text object
        vector.put(new Text(val), tmpInt);
    }

    // update the IntWritable wrapped int value
    tmpInt.set(tmpInt.get() + 1);

    // Note: you don't need to re-insert the IntWritable into the map
}
于 2012-05-14T18:25:27.060 回答