java - 如何在 Mapreduce 程序中遍历 Text 值的迭代器两次？

Question

在我的 MapReduce 程序中，我有一个 reducer 函数，它计算文本值迭代器中的项目数，然后对于迭代器中的每个项目，将项目输出为键，将计数输出为值。因此我需要使用迭代器两次。但是一旦迭代器到达终点，我就无法从第一个迭代器开始迭代。我该如何解决这个问题？我为我的 reduce 函数尝试了以下代码:

   public static class ReduceA extends MapReduceBase implements Reducer<Text, Text, Text, Text>
{

        public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text>output, Reporter reporter) throws IOException 
        {
            Text t;
            int count =0;                
            String[] attr = key.toString().split(",");      

           while(values.hasNext())               
            {
                values.next();
                count++;

            }

           //Maybe i need to reset my iterator here and start from the beginning but how do i do it?

           String v=Integer.toString(count);
           while(values.hasNext())               
            {
                t=values.next();

                output.collect(t,new Text(v));
            }
        }  
      }

上面的代码产生了空结果。我曾尝试将迭代器的值插入列表中，但由于我需要处理许多 GB 的数据，所以我在使用列表时遇到 Java 堆空间错误。请帮助我修改我的代码，以便我可以遍历迭代器两次。

Answer 1

您始终可以采用简单的方式来做到这一点:声明一个 List 并在您第一次迭代时缓存该值。因此，您可以遍历 List 并写出输出。你应该有类似这样的东西:

public static class ReduceA extends MapReduceBase implements
    Reducer<Text, Text, Text, Text> {

public void reduce(Text key, Iterator<Text> values,
        OutputCollector<Text, Text> output, Reporter reporter)
        throws IOException {
    Text t;
    int count = 0;
    String[] attr = key.toString().split(",");
    List<Text> cache = new ArrayList<Text>();

    while (values.hasNext()) {
        cache.add(values.next());
        count++;

    }

    // Maybe i need to reset my iterator here and start from the beginning
    // but how do i do it?

    String v = Integer.toString(count);
    for (Text text : cache) {
        output.collect(text, new Text(v));
    }
}
}