具有递归映射的 Hadoop MapReduce

Question

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

在 Hadoop 框架上实现此功能的最佳方式是什么？

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

Answer 1

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

所以，让我们举一个简单的（线性）示例。您有一个带有{I1, I2, I3, I4} 的一维空间。邻居将仅表示“上一个或下一个元素”：I1 是 I2 的邻居，但不是 I3。

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

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

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

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

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

在你的 reducer 中，像这样迭代所有邻居的值：

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

在你的map阶段，为每个邻居输出一个key，然后在reduce中求和。伪代码：

function map(index, count):
  for neighbor in neighbors(index):
     emit(neighbor, count)

function reduce(index, counts):
  total = sum(counts)
  emit(index, total)

它不是“递归的”，但如果我理解正确，它应该可以解决您的具体问题。