java - 在Java MapReduce中，Reducer的可迭代值似乎不一致

Question

我的reduce函数中有以下代码。当我尝试使用CollectionUtils.addAll创建浅拷贝时，该副本不成功。所有项目都将具有LAST项目的引用，而不是迭代器中的其他项目。

这是我的Reducer中的代码:

public void reduce(Text key, Iterable<ArrayListWritable<Writable>> values, Context context)
    throws IOException, InterruptedException {
    ArrayList<ArrayListWritable<Writable>> listOfWordPairs = new ArrayList<ArrayListWritable<Writable>>();

    // CollectionUtils.addAll(listOfWordPairs, values.iterator());
    // listOfWordPairs seems to all be the last item in the iterator

    Iterator<ArrayListWritable<Writable>> iter = values.iterator();

    // Manually do the copy
    while (iter.hasNext()) {
        // listOfWordPairs.add(iter.next()); 
        //Same behaviour as CollectionUtils.addAll()

        listOfWordPairs.add(new ArrayListWritable<Writable>(iter.next())); 
        //Only working way to do it -> deep copy :(
        }
    }

有人知道为什么会这样吗？我可以看到，如果MR以这种方式实现，则可以节省相当大的内存，但是似乎有些神奇的事情可以实现。我是MR新手，所以希望这个问题不会太愚蠢...

这是我对感兴趣的人的MAP代码

@Override
        public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
          Map<String, HMapStFW> stripes = new HashMap<>();

          List<String> tokens = Tokenizer.tokenize(value.toString());

          if (tokens.size() < 2) return;
          context.getCounter(StripesPmiEnums.TOTALENTRIES).increment(tokens.size());

          for (int i = 0; i < tokens.size() && i<40; i++) {
            for (int j = 0;j<tokens.size() && j<40;j++){
                if (j == i)
                    continue;
                //Make Stripe if doesn't exist
                if (!stripes.containsKey(tokens.get(i))){
                    HMapStFW newStripe = new HMapStFW();
                    stripes.put(tokens.get(i), newStripe);
                }

                HMapStFW stripe = stripes.get(tokens.get(i));
                if (stripe.containsKey(tokens.get(j))){
                    stripe.put(tokens.get(j), stripe.get(tokens.get(j))+1.0f);
                }else{
                    stripe.put(tokens.get(j), 1.0f);
                }
            }
          }

          for (String word1 : stripes.keySet()) {
            TEXT.set(word1);
            context.write(TEXT, stripes.get(word1));
          }
        }

ArrayListWritable也可以在这里使用
https://github.com/lintool/tools/blob/master/lintools-datatypes/src/main/java/tl/lin/data/array/ArrayListWritable.java

Answer 1

这是因为迭代器在 reducer 中的工作方式不同。简短的答案，您必须在迭代迭代器时克隆对象

while (iter.hasNext()) {
    //this is correct
    listOfWordPairs.add(new ArrayListWritable<Writable>(iter.next())); 

    }
}

看一下下面的链接，很好的解释了

https://cornercases.wordpress.com/2011/08/18/hadoop-object-reuse-pitfall-all-my-reducer-values-are-the-same/