使用 Hadoop MapReduce 排序字数

Question

我对 MapReduce 非常陌生，并且完成了 Hadoop 字数统计示例。

在该示例中，它生成未排序的字数文件（带有键值对）。那么是否可以通过将另一个 MapReduce 任务与之前的任务结合起来，按单词出现次数对其进行排序？

Answer 1

在简单的字数统计映射缩减程序中，我们得到的输出是按单词排序的。示例输出可以是：
苹果1
男孩 30
猫 2
青蛙20
斑马1
如果您希望输出根据单词出现次数进行排序，即采用以下格式
1 个苹果
1 斑马
2 猫
20 青蛙
30 男孩
您可以使用下面的映射器和化简器创建另一个 MR 程序，其中输入将是从简单字数统计程序获得的输出。

class Map1 extends MapReduceBase implements Mapper<Object, Text, IntWritable, Text>
{
    public void map(Object key, Text value, OutputCollector<IntWritable, Text> collector, Reporter arg3) throws IOException 
    {
        String line = value.toString();
        StringTokenizer stringTokenizer = new StringTokenizer(line);
        {
            int number = 999; 
            String word = "empty";

            if(stringTokenizer.hasMoreTokens())
            {
                String str0= stringTokenizer.nextToken();
                word = str0.trim();
            }

            if(stringTokenizer.hasMoreElements())
            {
                String str1 = stringTokenizer.nextToken();
                number = Integer.parseInt(str1.trim());
            }

            collector.collect(new IntWritable(number), new Text(word));
        }

    }

}


class Reduce1 extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text>
{
    public void reduce(IntWritable key, Iterator<Text> values, OutputCollector<IntWritable, Text> arg2, Reporter arg3) throws IOException
    {
        while((values.hasNext()))
        {
            arg2.collect(key, values.next());
        }

    }

}

Answer 2

Hadoop MapReduce wordcount 示例的输出按键排序。所以输出应该按字母顺序排列。

使用 Hadoop，您可以创建自己的关键对象来实现 WritableComparable 接口，从而允许您重写compareTo 方法。这允许您控制排序顺序。

要创建按出现次数排序的输出，您可能需要添加另一个 MapReduce 作业来处理第一个作业的输出，正如您所说的那样。第二项工作将非常简单，甚至可能不需要减少阶段。您只需要实现自己的 Writable 键对象来包装单词及其频率。自定义可写看起来像这样：

 public class MyWritableComparable implements WritableComparable {
       // Some data
       private int counter;
       private long timestamp;

       public void write(DataOutput out) throws IOException {
         out.writeInt(counter);
         out.writeLong(timestamp);
       }

       public void readFields(DataInput in) throws IOException {
         counter = in.readInt();
         timestamp = in.readLong();
       }

       public int compareTo(MyWritableComparable w) {
         int thisValue = this.value;
         int thatValue = ((IntWritable)o).value;
         return (thisValue < thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
       }
     }

我从 此处。

您可能还应该重写 hashCode、equals 和 toString。

Answer 3

在 Hadoop 中，排序是在 Map 和 Reduce 阶段之间完成的。按单词出现次数排序的一种方法是使用不对任何内容进行分组的自定义组比较器；因此，每次调用reduce都只是一个键和一个值。

public class Program {
   public static void main( String[] args) {

      conf.setOutputKeyClass( IntWritable.class);
      conf.setOutputValueClass( Text.clss);
      conf.setMapperClass( Map.class);
      conf.setReducerClass( IdentityReducer.class);
      conf.setOutputValueGroupingComparator( GroupComparator.class);   
      conf.setNumReduceTasks( 1);
      JobClient.runJob( conf);
   }
}

public class Map extends MapReduceBase implements Mapper<Text,IntWritable,IntWritable,Text> {

   public void map( Text key, IntWritable value, OutputCollector<IntWritable,Text>, Reporter reporter) {
       output.collect( value, key);
   }
}

public class GroupComaprator extends WritableComparator {
    protected GroupComparator() {
        super( IntWritable.class, true);
    }

    public int compare( WritableComparable w1, WritableComparable w2) {
        return -1;
    }
}

Answer 4

正如您所说，一种可能性是编写两个作业来完成此操作。
第一份工作：
简单的字数统计示例

第二份工作：
进行排序部分。

伪代码可以是：

注意：第一个作业生成的输出文件将作为第二个作业的输入

    Mapper2(String _key, Intwritable _value){
    //just reverse the position of _value and _key. This is useful because reducer will get the output in the sorted and shuffled manner.
    emit(_value,_key);
    }

    Reduce2(IntWritable valueofMapper2,Iterable<String> keysofMapper2){
//At the reducer side, all the keys that have the same count are merged together.
        for each K in keysofMapper2{
        emit(K,valueofMapper2); //This will sort in ascending order.
        }

    }

您还可以按降序排序，为此编写一个单独的比较器类是可行的会成功的。
在作业中包含比较器，如下所示：

Job.setComparatorclass(Comparator.class);

该比较器将在发送到减速器端之前按降序对值进行排序。因此，在减速器上，您只需发出值。