Hadoop / MapReduce - 优化“前 N”字数列表映射减少作业

Question

我正在做一些类似于标准MapReduce示例的事情——字数统计，但是有所改变，我只希望得到前N个结果。

假设我在HDFS有一个非常大的文本数据集。有大量的例子展示了如何构建一个Hadoop MapReduce作业，为你提供文本中每个单词的字数。例如，如果我的语料库是:

“这是对测试数据的检验，也是检验这一点的好方法”

来自标准 MapReduce 字数统计作业的结果集为：

测试：3、a：2、this：2、is：1等。

但是，如果我只想获得在我的整个数据集中使用的前3个单词呢？

我仍然可以运行完全相同的标准MapReduce单词计数作业，然后在它准备好并输出每个单词的计数后，只获取前3个结果，但这似乎有点低效，因为在洗牌阶段需要移动大量数据。

我想的是，如果这个样本足够大，并且数据在HDFS中是随机且分布良好的，那么每个映射器不需要将其所有字数发送到Reducers，而只需要将一些顶部数据发送到Reducers。所以如果一个映射器有这样的：

a： 8234，人：5422，人：4352，…更多的词…，rareword:1，古怪词：1，等等。

然后，我想做的只是将每个映射器的前100个左右的单词发送到Reducer阶段——因为当所有的事情都结束时,“稀有单词”突然出现在前3名的可能性很小。这似乎可以节省带宽和Reducer处理时间。

这可以在组合器阶段完成吗？这种在洗牌阶段之前的优化通常会完成吗？

Answer 1

引用托马斯的话

要获得Top N，您只需将本地HashMultiset中的Top N写入输出收集器，并以正常方式在减少端聚合结果。这也为您节省了大量网络带宽，唯一的缺点是您需要在清理方法中对字数元组进行排序。

如果您只在本地HashMultiset中写入前N个元素，则可能会丢失元素的计数，如果从本地Hash多重集合传递，该元素可能会成为总的前10个元素之一。

例如，将以下格式视为三个映射，即MapName: elementName，elemenntcount:

地图 A ： Ele1，4 ： Ele2，5 ： Ele3，5 ： Ele4，2

地图 B ： Ele1，1 ： Ele5，7 ： Ele6， 3 ： Ele7，6

地图C:Ele5,4:Ele8,3:Ele1,1:Ele9,3

现在，如果我们考虑每个映射器的前3个，我们将错过元素“Ele1”，其总数应该是6，但由于我们正在计算每个映射器的前3个，我们看到“Ele1”的总计数为4。

我希望这是有道理的。请让我知道你对此的看法。

Answer 2

这是一个非常好的问题，因为您已经发现了Hadoop的单词计数示例的低效性。

优化问题的技巧如下：

在本地地图阶段执行基于HashMap的分组，您也可以使用组合器。这看起来像这样，我正在使用Guava的HashMultiSet，它促进了一个很好的计数机制。

    public static class WordFrequencyMapper extends
      Mapper<LongWritable, Text, Text, LongWritable> {

    private final HashMultiset<String> wordCountSet = HashMultiset.create();

    @Override
    protected void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {

      String[] tokens = value.toString().split("\\s+");
      for (String token : tokens) {
        wordCountSet.add(token);
      }
    }

在清理阶段，您会发出结果:

@Override
protected void cleanup(Context context) throws IOException,
    InterruptedException {
  Text key = new Text();
  LongWritable value = new LongWritable();
  for (Entry<String> entry : wordCountSet.entrySet()) {
    key.set(entry.getElement());
    value.set(entry.getCount());
    context.write(key, value);
  }
}

因此，您将单词分组到本地工作块中，从而通过使用一位RAM来减少网络使用。您也可以使用＜code＞组合器

要获得前N名，只需将本地< code>HashMultiset中的前N名写入输出收集器，并在reduce端以正常方式聚合结果。这也为您节省了大量的网络带宽，唯一的缺点是您需要在清理方法中对字数元组进行排序。

部分代码可能如下所示:

  Set<String> elementSet = wordCountSet.elementSet();
  String[] array = elementSet.toArray(new String[elementSet.size()]);
  Arrays.sort(array, new Comparator<String>() {

    @Override
    public int compare(String o1, String o2) {
      // sort descending
      return Long.compare(wordCountSet.count(o2), wordCountSet.count(o1));
    }

  });
  Text key = new Text();
  LongWritable value = new LongWritable();
  // just emit the first n records
  for(int i = 0; i < N, i++){
    key.set(array[i]);
    value.set(wordCountSet.count(array[i]));
    context.write(key, value);
  }

我希望你能领会在本地尽可能多地使用这个词的要点，然后只汇总前N个中的前N个；)