java - 如何将 MapWritable 与 SequenceFile 一起使用？ Hadoop

Question

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

公共(public)静态类 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);

        }
    }

和映射器:

公共(public)静态类 TokenizerMapper 扩展映射器{

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

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

这就是你的问题 - hadoop 重用了 val Text 对象，所以在调用 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
}