Hadoop MapReduce与递归映射

Question

我需要在Java中做一个MapReduce应用程序，它需要自动递归，这意味着对于处理的每一行输入文件，它必须检查输入/映射条目的所有行是否存在条件，并由函数验证。或者，换句话说，Reducer 应该调用/读取收到的每对（键、值）的所有 Map。

在Hadoop框架上实现它的最佳方法是什么？

我可以通过读取输入n次或将输入加载到哈希图中以编程方式做到这一点，但我认为可以在MapReduce范式中完成所有这些操作。 感谢您的任何帮助/提示！ 编辑：更多细节，我有（由于其他工作）问题空间的分区列表（索引，计数），并希望作为输出（索引，sumOfNearestNeighborsCounts），因此对于每个索引，我想再次访问地图，并且对于每个NearestNeighbor索引的总和，发生次数。 （另见Costi Ciudatu评论）

Answer 1

对于每个索引键，您需要发出所有可能的邻居索引（您应该能够以数学方式生成）。

因此，让我们举一个简单的（线性）示例。你有一个一维空间，{I1, I2, I3, I4}.邻居只是表示"上一个或下一个元素"：I1 是 I2 的邻居，但不是 I3。

对于每个进入映射器的索引，为该索引的每个可能的邻居（包括它自己！）发出一个键 - 我们将定义每个索引都是自身的可能邻居，但计数具有特殊且荒谬的负值，我将解释原因）：

<I1, count(I1)> -> <I0, count(I1)>
                -> <I1, -1>
                -> <I2, count(I1)>
<I2, count(I2)> -> <I1, count(I2)>
                -> <I2, -1>
                -> <I3, count(I2)>

现在，在化简器中，您将获得每个键的以下值：

I0: [ count(I1) ]
I1: [ count(I2), -1 ]
I2: [ count(I1), -1, count(I3) ]
...

在化简器中，像这样迭代邻居的所有值：

boolean doesExist = false;
int sum = 0;
for (IntWritable value : values) {
    int count = value.get();
    if (count < 0) {
        doesExist = true;
    } else {
        sum += count;
    }
}
if (doesExist) {
    context.write(key, new IntWritable(sum));
}

这样，您将排除（在上面的示例中）I0 和 I4，它们不存在，并且它们的列表中没有负值。

---

现在，为了更接近您的用例，如果您在迭代期间也需要实际索引值（而不仅仅是所有邻居的计数），您可以执行以下操作：

不要从映射器发出简单的数字，而是输出一些包含索引及其计数的包装器bean。这样，您将能够根据某些业务约束或其他限制排除一些邻居，但您将始终只使用每个给定索引的（可能的）邻居列表，而不是整个输入集：

<I1, count(I1)> -> <I0, {I1, count(I1)}>
                -> <I1, {I1, count(I1)}>
                -> <I2, {I1, count(I1)}>
... and so on

现在，在减速器中，您将获得：

I0: [ {I1, count(I1)} ]
I1: [ {I1, count(I1)}, {I2, count(I2)} ]
I2: [ {I1, count(I1)}, {I2, count(I2)}, {I3, count(I3)} ]

正如你所注意到的，你不再需要人工-1计数，至于doesExist测试，你现在可以检查值列表中的任何包装器bean是否与键索引具有相同的索引。

即使可能的邻居数量随着维度的数量呈指数增长（正如你已经提到的），我想说这种方法仍然比读取每个键/值对的整个输入要好得多，而且它更适合map/reduce范式。

Answer 2

在映射阶段，为每个邻居输出一个键，然后在归约中求和。伪代码：

function map(index, count):
  for neighbor in neighbors(index):
     emit(neighbor, count)
function reduce(index, counts):
  total = sum(counts)
  emit(index, total)

它不是"递归的"，但如果我理解正确，它应该可以解决您的特定问题。