mapreduce 排序

本帖最后由 hery 于 2015-11-18 18:34 编辑

[mw_shl_code=applescript,true]package cn.edu.ruc.cloudcomputing.book.chapter05;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.mapreduce.Partitioner;

public class Sort {

//map将输入中的value转化成IntWritable类型，作为输出的key
  public static class Map extends Mapper<Object, Text, IntWritable, IntWritable>{

private static IntWritable data = new IntWritable();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
   String line = value.toString();
   data.set(Integer.parseInt(line));
   context.write(data, new IntWritable(1));
}
  }
//reduce将输入的key复制到输出的value，然后根据输入的
//value-list中元素的个数决定key的输出次数
//用全局linenum来代表key的位次
  public static class Reduce extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {
private static IntWritable linenum = new IntWritable(1);

public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
   for (IntWritable val : values) {
      context.write(linenum , key);
      linenum = new IntWritable(linenum.get() + 1);
   }
}
  }
//自定义Partition函数，此函数根据输入数据的最大值和MapReduce框架中
//Partition的数量获取将输入数据按照大小分块的边界，然后根据输入数值和
//边界的关系返回对应的Partition ID
public static class Partition extends Partitioner <IntWritable,IntWritable> {
            @Override
      public int getPartition(IntWritable key, IntWritable value, int numPartitions) {

      int Maxnumber = 65223;
      int bound = Maxnumber/numPartitions + 1;
      int keynumber = key.get();
      for(int i = 0; i < numPartitions; i++){
            if(keynumber < bound*i && keynumber >= bound*(i-1))
                     return i-1;
            }
            return -1;

            }
  }

  public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
   System.err.println("Usage: wordcount <in> <out>");
   System.exit(2);
}
Job job = new Job(conf, "Sort");
job.setJarByClass(Sort.class);
job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
job.setPartitionerClass(Partition.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}
[/mw_shl_code]
求大神给讲解一下，那个名次是如何求出来的？我对于求名次有点不明白，谢谢

bioger_hit · 发表于 2015-11-18 19:31:54

这里其实楼主只要理解普通的mapreduce，这里的，map函数和reduce函数，几乎没有任何内容，跟其它mapreduce函数都是一样的。
map只起到了类型转换
reduce函数就是一个简单的循环输出，当然这些只是单纯从代码的角度来解。
如果从mapreduce的原理来理解，这里就复杂了。
楼主暂时从代码角度看，这里更容易些。

这里面最关键的内容是分区函数。
分区函数什么？恐怕楼主还不明白。
分区其实就是比如我们对大豆分类，如果我们按照颜色来分，分为黄豆，绿豆，黑豆
如果我们按照大小来分，那么分为大豆，中豆，小豆。

那么颜色和大小就类似于分区函数。

理解了分区函数，我们就知道了这些数字是如何排序的了。

首先这些数字会到map里进行分割，当然这里无需分割，然后通过分区函数，会对输入的数据进行分类，分类之后，reduce输出即可。

楼主可以尝试把分区函数改变或则注释掉，肯定结果会发生改变

hery · 发表于 2015-11-27 16:47:45

bioger_hit 发表于 2015-11-18 19:31
这里其实楼主只要理解普通的mapreduce，这里的，map函数和reduce函数，几乎没有任何内容，跟其它mapred ...

分区函数我明白，它在这里的作用就是把数字进行了分类，比如1-10之间的数作为一组，11-20的做为一组。。。。90-100的作为一组，但是在reduce函数中的value-list,比如一个key:50只有一个，那么value-list的值为<1>,所以它只循环一次，那么循环一次的话，如何得出实际排名为3呢

linbowei · 发表于 2015-11-30 17:20:10

15/11/30 17:15:21 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/11/30 17:15:21 INFO input.FileInputFormat: Total input paths to process : 2
15/11/30 17:15:21 WARN snappy.LoadSnappy: Snappy native library not loaded
15/11/30 17:15:22 INFO mapred.JobClient: Running job: job_local_0001
15/11/30 17:15:22 INFO mapred.Task: Using ResourceCalculatorPlugin : null
15/11/30 17:15:22 INFO mapred.MapTask: io.sort.mb = 100
15/11/30 17:15:22 INFO mapred.MapTask: data buffer = 79691776/99614720
15/11/30 17:15:22 INFO mapred.MapTask: record buffer = 262144/327680
15/11/30 17:15:22 WARN mapred.LocalJobRunner: job_local_0001
java.io.IOException: Illegal partition for 1 (-1)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1073)
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:691)
at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
at my.Sort$Map.map(Sort.java:27)
at my.Sort$Map.map(Sort.java:1)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:370)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:212)
15/11/30 17:15:23 INFO mapred.JobClient: map 0% reduce 0%
15/11/30 17:15:23 INFO mapred.JobClient: Job complete: job_local_0001
15/11/30 17:15:23 INFO mapred.JobClient: Counters: 0
在我的虚拟机跑，会报这样的错误