对值迭代两次 (MapReduce)答案

【问题标题】：Iterate twice on values (MapReduce)对值迭代两次 (MapReduce)
【发布时间】：2011-08-31 23:02:53
【问题描述】：

我收到一个迭代器作为参数，我想对值进行两次迭代。

public void reduce(Pair<String,String> key, Iterator<IntWritable> values,
                   Context context)

有可能吗？如何？签名是由我正在使用的框架（即 Hadoop）强加的。

-- 编辑--
最后，reduce 方法的真正签名是iterable。我被这个 wiki page 误导了（这实际上是我发现的唯一一个未弃用（但错误）的 wordcount 示例）。

【问题讨论】：

我想我可以先将每个值存储在一个容器中并在其上迭代两次，但是......说真的......我希望有更好的东西
出于好奇，需要迭代两次吗？
无论你做什么，只是不要在迭代器上迭代两次
@Matt Ball：在很多情况下，您可能想要对一个集合进行两次迭代。以“多数选举”问题为例，当您必须知道集合 C 中是否存在超过 size(C)/2 次的元素 E 时。您需要首先使用cs.utexas.edu/~moore/best-ideas/mjrty/index.html 对元素进行完整迭代，仅当此类元素存在时才会给出正确答案，然后在实际检查“猜测的主要元素”是否真的是主要元素时进行第二次遍历.
@log0：你解决了答案吗？当我迭代第二个循环时。我的 cahe 列表被覆盖了。

标签： java iterator hadoop

【解决方案1】：

重用给定的迭代器，不。

但是您可以在首先迭代它们然后迭代构造的 ArrayList 时将值保存在 ArrayList 中，当然（或者您可以通过使用一些花哨的 Collection 方法直接构建它，然后迭代直接在 ArrayList 上两次。这是口味问题）。

不管怎样，你确定传递一个迭代器是一件好事吗？迭代器仅用于对集合进行线性扫描，这就是它们不公开“rewind”方法的原因。

您应该传递一些不同的东西，例如 Collection<T> 或 Iterable<T>，正如其他答案中已经建议的那样。

【讨论】：

好的，所以我想到的解决方案......（正如我在评论中所说）。否则我认为我对签名无能为力。它是由 Hadoop 框架（我正在使用的）强加的。

【解决方案2】：

迭代器只能遍历一次。一些迭代器类型是可克隆的，你也许可以在遍历之前克隆它，但这不是一般情况。

你应该让你的函数取一个Iterable，如果你能做到的话。

【讨论】：

【解决方案3】：

如果你想再次迭代，我们必须缓存来自迭代器的值。至少我们可以将第一次迭代和缓存结合起来：

Iterator<IntWritable> it = getIterator();
List<IntWritable> cache = new ArrayList<IntWritable>();

// first loop and caching
while (it.hasNext()) {
   IntWritable value = it.next();
   doSomethingWithValue();
   cache.add(value);
}

// second loop
for(IntWritable value:cache) {
   doSomethingElseThatCantBeDoneInFirstLoop(value);
}

（只是用代码添加答案，知道您在自己的评论中提到了这个解决方案；））

为什么没有缓存是不可能的：Iterator 是实现接口的东西，并且没有一个要求，Iterator 对象实际上存储值。重复两次，要么必须重置迭代器（不可能），要么克隆它（再次：不可能）。

举一个克隆/重置没有任何意义的迭代器示例：

public class Randoms implements Iterator<Double> {

  private int counter = 10;

  @Override 
  public boolean hasNext() { 
     return counter > 0; 
  }

  @Override 
  public boolean next() { 
     count--;
     return Math.random();        
  }      

  @Override 
  public boolean remove() { 
     throw new UnsupportedOperationException("delete not supported"); 
  }
}

【讨论】：

您应该将cache 的声明至少更改为Collection，这样您才能真正调用add。
正确。不记得为什么我将cache 声明为Iterable。闻起来像复制粘贴神器;)
@Andreas_D：当我执行上述代码时，我的缓存列表被新值覆盖

【解决方案4】：

如果方法签名无法更改，那么我建议使用Apache Commons IteratorUtils 将 Iterator 转换为 ListIterator。考虑这个对值进行两次迭代的示例方法：

void iterateTwice(Iterator<String> it) {
    ListIterator<?> lit = IteratorUtils.toListIterator(it);
    System.out.println("Using ListIterator 1st pass");
    while(lit.hasNext())
        System.out.println(lit.next());

    // move the list iterator back to start
    while(lit.hasPrevious())
        lit.previous();

    System.out.println("Using ListIterator 2nd pass");
    while(lit.hasNext())
        System.out.println(lit.next());
}

使用上面这样的代码，我能够迭代值列表而不在我的代码中保存 List 元素的副本。

【讨论】：

但无论如何它都会这样做，因此内存利用率或其他方面没有区别......这只是节省两行代码的一种奇特方式。这是否证明导入库是合理的？
至少在我的情况下，出于某种原因，我的大多数应用程序已经将 apache commons 集合作为依赖项。 IMO 任何节省编写自己的本地代码的方法都是更好的（阅读更清晰）代码，但当然，您始终可以采用自己保存这些值的第一个建议。
嗯，我认为使用一些外部库会使您的代码更具可读性并不总是如此，因为不是每个人都对您正在使用的库有所了解。对于复杂或非常无聊的任务，使用库总是一个好主意，但对于像这样的“微任务”，你必须弄清楚“IteratorUtils.toListIterator()”是做什么的，当一个存储值的循环立即出现时可以理解的。不要误会我的意思，我非常喜欢 Apache Commons，但我认为我们应该使用（外部）库商品与简约。
这正是我的观点，如果它是一些不为人知的闻所未闻的库类型，我们必须在使用之前验证所有内容。但是“Apache commons”是其公共套件中使用最广泛的库之一。正如我所说，几乎我所有的应用程序都已经在使用它，所以它并不是依赖项的新添加。
@anubhava：它对我来说是部分工作。我还需要两次迭代。但是当我通过应用你的代码进行检查时。在第一次通过时，我能够正确地获得所有值。但是对于 2 nd pass 我只重复获取第一个元素。我们是否能够在两次传递中获得相同的值

【解决方案5】：

不幸的是，如果不缓存 Andreas_D 的答案中的值，这是不可能的。

即使使用Reducer 接收Iterable 而不是Iterator 的新API，您也不能迭代两次。尝试以下方法非常诱人：

for (IntWritable value : values) {
    // first loop
}

for (IntWritable value : values) {
    // second loop
}

但这实际上行不通。您从该Iterable 的iterator() 方法收到的Iterator 是特殊的。这些值可能并不都在内存中； Hadoop 可能正在从磁盘流式传输它们。它们并没有真正得到Collection 的支持，因此允许多次迭代并非易事。

您可以在Reducer 和ReduceContext 代码中亲自看到这一点。

在某种Collection 中缓存值可能是最简单的答案，但如果您在大型数据集上操作，您可以轻松地破坏堆。如果您能就您的问题向我们提供更多细节，我们或许可以帮助您找到不涉及多次迭代的解决方案。

【讨论】：

谢谢你，很高兴知道。

【解决方案6】：

如果我们尝试在 Reducer 中迭代两次，如下所示

ListIterator<DoubleWritable> lit = IteratorUtils.toListIterator(it);
System.out.println("Using ListIterator 1st pass");
while(lit.hasNext())
    System.out.println(lit.next());

// move the list iterator back to start
while(lit.hasPrevious())
    lit.previous();

System.out.println("Using ListIterator 2nd pass");
while(lit.hasNext())
    System.out.println(lit.next());

我们只会输出为

Using ListIterator 1st pass
5.3
4.9
5.3
4.6
4.6
Using ListIterator 2nd pass
5.3
5.3
5.3
5.3
5.3

为了以正确的方式获取它，我们应该像这样循环：

ArrayList<DoubleWritable> cache = new ArrayList<DoubleWritable>();
 for (DoubleWritable aNum : values) {
    System.out.println("first iteration: " + aNum);
    DoubleWritable writable = new DoubleWritable();
    writable.set(aNum.get());
    cache.add(writable);
 }
 int size = cache.size();
 for (int i = 0; i < size; ++i) {
     System.out.println("second iteration: " + cache.get(i));
  }

输出

first iteration: 5.3
first iteration: 4.9
first iteration: 5.3
first iteration: 4.6
first iteration: 4.6
second iteration: 5.3
second iteration: 4.9
second iteration: 5.3
second iteration: 4.6
second iteration: 4.6

【讨论】：

+1 但它不适合大型数据集，因为我们正在创建同一个列表的副本

【解决方案7】：

试试这个：

    ListIterator it = list.listIterator();

    while(it.hasNext()){

        while(it.hasNext()){
            System.out.println("back " + it.next() +" "); 
        }
        while(it.hasPrevious()){
            it.previous();
        }
    }

【讨论】：

@Shevliaskovic，这似乎是不言而喻的：代码向前传递列表，然后转身向后传递第二次。

【解决方案8】：

如果你想随时更改值，我想最好使用 listIterator 然后使用它的 set() 方法。

ListIterator lit = list.listIterator();
while(lit.hasNext()){
   String elem = (String) lit.next();
   System.out.println(elem);
   lit.set(elem+" modified");
}
lit = null; 
lit = list.listIterator();
while(lit.hasNext()){
   System.out.println(lit.next());
}

我没有调用 .previous()，而是在同一个列表迭代器对象上获取 .listIterator() 的另一个实例。

【讨论】：

【解决方案9】：

你可以这样做

MarkableIterator<Text> mitr = new MarkableIterator<Text>(values.iterator());
mitr.mark();
while (mitr.hasNext()) 
{
//do your work
}
mitr.reset();
while(mitr.hasNext()) 
{
//again do your work
}

【讨论】：

【解决方案10】：

经过多次尝试和错误，我找到了解决方案。

声明一个新集合（比如cache）（链表或Arraylist或任何其他）
在第一次迭代中，分配当前迭代器，如下例：
```
cache.add(new Text(current.get()))  
```

遍历缓存：

for (Text count : counts) {
    //counts is iterable object of Type Text
    cache.add(new Text(count.getBytes()));
}
for(Text value:cache) {
    // your logic..
}

【讨论】：

【解决方案11】：

注意：如果您使用缓存列表来缓存项目，您应该先克隆该项目，然后再添加到缓存中。否则你会发现缓存中的所有项目都是一样的。

这种情况是由于MapReduce的内存优化造成的，在reduce方法中，Iterable重用了item实例，详细可以找here

【讨论】：