第一次迭代時Hadoop迭代器跳過方法調用

Question

我有一個Map Reduce程序，並且在Reducer類中，我的方法在第一次迭代中沒有被調用。我想要實現的是在迭代器的每兩個連續值之間生成一些新行。 （對如：（1,2），（2,3），（3,4）...）。我錯過了什麼？我還測試了我有我需要的對，它看起來很好，但似乎我的方法不是第一對調用。 generate（） - 將在每兩個連續行之間生成新行（填滿時間間隙）

input: X, Y, 00:00:00, 908 X, Y, 00:00:05, 122 X, Y, 00:00:07, 123

desired output: X, Y, 00:00:00, 908 X, Y, 00:00:01, 908 X, Y, 00:00:02, 908 X, Y, 00:00:03, 908 X, Y, 00:00:04, 908 X, Y, 00:00:05, 122 X, Y, 00:00:06, 122 X, Y, 00:00:07, 123

Iterator<MyType> iterator = values.iterator(); 
if (!iterator.hasNext()) return; 
first = iterator.next(); 
while (iterator.hasNext()) { 
    nr++; 
    first.setStatus(nr); 
    context.write(nullWritable, first); 
    second = iterator.next(); 
    List<MyType> newValues = generate(first, second, context); 
    for (MyType mt : newValues) { 
      mt.setStatus(nr); 
      context.write(nullWritable, mt); 
    } 
    second.setStatus(nr); 
    context.write(nullWritable, second); 
    first = new InterpolationModelWritable(); 
    first.setX(second.getX()); 
    first.setY(second.getY()); 
    first.setZ(second.getZ()); 
    first.setTag(second.getTag()); 

} 

actual result: X, Y, 00:00:00, 908 X, Y, 00:00:05, 122 X, Y, 00:00:06, 122 X, Y, 00:00:07, 123

Answer 1

與您的代碼的問題是，你落入Hadoop的對象重用陷阱。重要的是要記住，reduce中的值迭代器在每次調用next()時都不會返回一個新對象，它將重用同一個對象。

現在我們知道我們可以看看你的代碼並找出什麼是錯的。使用你的邏輯（但作爲一個最小的工作示例），我們可以看到通常是它會工作。

@Test 
public void noResuseIteratorTest() { 

    List<String> values = new ArrayList<>(); 
    values.add("a"); 
    values.add("b"); 
    values.add("c"); 
    values.add("d"); 

    String first; 
    String second; 

    Iterator<String> iterator = values.iterator(); 
    if (!iterator.hasNext()) return; 
    first = iterator.next(); 
    while (iterator.hasNext()) { 
     second = iterator.next(); 
     System.out.println("Out: " + first + " - " + second); 
     first = second; 
    } 
} 

輸出：一個 - B
缺貨：乙 - Ç
缺貨：C - d

然而，在減少的Hadoop方法返回的值是相同的對象。該測試說明問題：

@Test 
public void reuseIteratorTest() { 

    class MyList implements Iterable<Text> { 

     private List<String> myList = new ArrayList<>(); 
     private Text reused = new Text(); 

     public MyList() { 
      myList.add("a"); 
      myList.add("b"); 
      myList.add("c"); 
      myList.add("d"); 
     } 

     @Override 
     public Iterator<Text> iterator() { 
      return new Iterator<Text>() { 

       private final Iterator<String> iter = myList.iterator(); 

       @Override 
       public boolean hasNext() { 
        return iter.hasNext(); 
       } 

       @Override 
       public Text next() { 
        // We reuse the Text object here! 
        reused.set(iter.next()); 
        return reused; 
       } 

       @Override 
       public void remove() { 
        throw new UnsupportedOperationException(""); 
       } 
      }; 
     } 
    } 

    MyList myList = new MyList(); 
    Text first; 
    Text second; 

    Iterator<Text> iterator = myList.iterator(); 
    if (!iterator.hasNext()) return; 
    first = iterator.next(); 
    while (iterator.hasNext()) { 
     second = iterator.next(); 
     System.out.println("Out: " + first + " - " + second); 
     first = new Text(); 
     first.set(second); 
    } 
} 

出：乙 - B
出：乙 - ç
出：C - d

所以，簡單地做second = iterator.next();不夠好。在第一次迭代中，first和second都指向同一個對象。

要解決此問題，您需要將迭代器值的內容複製到對象中，而不是僅指向同一對象。使用文本對象爲例，在固定的版本會是什麼樣子：

@Test 
public void reuseFixedIteratorTest() { 

    // MyList class from above goes here 

    MyList myList = new MyList(); 
    Text first = new Text(); 
    Text second = new Text(); 

    Iterator<Text> iterator = myList.iterator(); 
    if (!iterator.hasNext()) return; 
    first.set(iterator.next()); 
    while (iterator.hasNext()) { 
     second.set(iterator.next()); 
     System.out.println("Out: " + first + " - " + second); 
     first.set(second); 
    } 
} 

輸出：一個 - B
出：乙 - ç
出：C - d