java - 在 Hadoop 中并行化 Ruby reducer？

Question

Ruby 中的一个简单的 wordcount reducer 如下所示:

#!/usr/bin/env ruby
wordcount = Hash.new
STDIN.each_line do |line|
keyval = line.split("|")
wordcount[keyval[0]] = wordcount[keyval[0]].to_i+keyval[1].to_i
end

wordcount.each_pair do |word,count|
puts "#{word}|#{count}"
end

它在 STDIN 中获取所有映射器的中间值。不是来自特定的 key 。所以实际上只有一个缩减器(而不是每个单词或每组单词的缩减器)。

但是，在 Java 示例中，我看到这个接口(interface)获取键和值列表作为 inout。这意味着在 reduced 和 reducers 可以并行运行之前，中间映射值按键分组:

public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
            public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
              int sum = 0;
              while (values.hasNext()) {
                sum += values.next().get();
              }
              output.collect(key, new IntWritable(sum));
            }
          }

这是 Java 独有的功能吗？或者我可以使用 Ruby 通过 Hadoop Streaming 来实现吗？

Answer 1

Reducer 将始终并行运行，无论您是否使用流式传输(如果您没有看到这一点，请验证作业配置是否设置为允许多个 reduce 任务——请参阅集群中的 mapred.reduce.tasks或作业配置)。不同之处在于，当您使用 Java 而不是流式传输时，框架会为您更好地打包一些东西。

对于 Java，reduce 任务获取一个针对特定键的所有值的迭代器。如果您要对 reduce 任务中的 map 输出求和，这将使遍历值变得容易。在流式传输中，您实际上只是获得了键值对流。您被保证值将按键排序，并且对于给定的键，不会在 reduce 任务中拆分，但您需要的任何状态跟踪都由您决定。例如，在 Java 中，您的 map 输出以符号形式出现在您的 reducer 中

key1, {val1, val2, val3}key2, {val7, val8}

通过流式传输，您的输出看起来像

key1, val1 key 1、值 2 key 1，值 3键 2，值 7key2, val8

例如，要编写一个计算每个键的值总和的缩减器，您需要一个变量来存储您看到的最后一个键和一个变量来存储总和。每次读取新的键值对时，您都会执行以下操作:

1. 检查 key 是否与上一个 key 不同。
2. 如果是，输出你的 key 和当前总和，并将总和重置为零。
3. 将当前值添加到您的总和并将最后一个键设置为当前键。

HTH.