java - Hadoop Iterator 在第一次迭代时跳过方法调用

Question

我有一个 Map Reduce 程序，在 Reducer 类中，我的方法在第一次迭代中没有被调用。我想要实现的是在迭代器的每 2 个连续值之间生成一些新行。 (对像:(1,2)，(2,3)，(3,4)......)。我错过了什么？而且我还测试了我有我需要的对，看起来不错，但似乎第一对没有调用我的方法.. generate() - 将在每 2 个连续行之间生成新行(填补时间间隔)

输入: X、Y、00:00:00、908 X、Y、00:00:05、122 X、Y、00:00:07、123

期望的输出: 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());

 }

实际结果: 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;
    }
}

Out: a - b
Out: b - c
Out: c - d

但是，在 hadoop reduce 方法中，返回值是同一个对象。这个测试演示了这个问题:

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

Out: b - b
Out: b - c
Out: c - d

因此，简单地执行 second = iterator.next(); 是不够的。在您的第一次迭代中，first 和 second 都指向同一个对象。

要解决此问题，您需要将迭代器值的内容复制到您的对象中，而不仅仅是指向同一个对象。以 Text 对象为例，固定版本如下所示:

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

Out: a - b
Out: b - c
Out: c - d