面接問題のMAPREDUCE

question:

フィールドは以下の通り(学年、クラス、科目、名前、点数)、それぞれgrade,class,subject,name,scoreに対応し、各学年の各クラスの各科目の最高点Top 10を求めてmapreduceで書く

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
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;

public class StudentScoreTopN {
 static class StudentScoreTopNMapper extends Mapper<LongWritable, Text, Text, DoubleWritable>{
  Text k = new Text();
  DoubleWritable v = new DoubleWritable() ;
  @Override
  protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DoubleWritable>.Context context)
    throws IOException, InterruptedException {
   //  
   String line = value.toString();
   String[] split = line.split(","); 
   //    key
   String str = split[0] + "--" + split[1] + "--" +  split[2] + "-->" + split[3] ;
   //   Double  value
   Double d = Double.parseDouble(split[4]);
   k.set(str);
   v.set(d);
   context.write(k, v); 
  }
 }

static class StudentScoreTopNReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
  Map<String, Double> map = new  HashMap<String , Double>();
  @Override
  protected void reduce(Text key, Iterable<DoubleWritable> iters,
    Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context)
    throws IOException, InterruptedException {
   // value
   for (DoubleWritable doubleWritable : iters) {
    double d = doubleWritable.get();
   // map  
    map.put(key.toString(), d);
   }
  }
  @Override
  protected void cleanup(Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context)
    throws IOException, InterruptedException { 
   // map list 
   Set<Entry<String, Double>> set = map.entrySet();
   ArrayList<Entry<String, Double>> list = new ArrayList<Entry<String, Double>>(set);
   // 
   Collections.sort(list , new Comparator<Entry<String, Double>>() {
    @Override
    public int compare(Entry<String, Double> o1, Entry<String, Double> o2) { 
     // , ,    
     String[] split1 = o1.getKey().split("-->");
     String[] split2 = o2.getKey().split("-->");
     int i = split1[0].compareTo(split2[0]);
     return i == 0 ?Double.compare(o2.getValue(), o1.getValue()) : i;
    }
   });
   //    Top10     10 
   for(int i = 0 ; i < Integer.min(10, list.size()) ; i++ ) {
    Text k = new Text();
    DoubleWritable v = new DoubleWritable();
    k.set(list.get(i).getKey());
    v.set(list.get(i).getValue());
    context.write(k, v);
   }
  }
 }

public static void main(String[] args) throws Exception {
  // 1        
  Configuration conf = new  Configuration();
  //2  Job
  Job job = Job.getInstance(conf);
  //3   maptask    reducetask 
  job.setMapperClass(StudentScoreTopNMapper.class);
  job.setReducerClass(StudentScoreTopNReducer.class);
  //   maptask kv 
  job.setMapOutputKeyClass(Text.class);
  job.setMapOutputValueClass(DoubleWritable.class);
  //   reducetask kv 
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(DoubleWritable.class);
  //4   
  FileInputFormat.setInputPaths(job, new Path("D://map//student//1.txt"));
  //5  
  FileOutputFormat.setOutputPath(job, new Path("D://map//student//stu4"));
  // Job   
  job.waitForCompletion(true);
 }
}

自分で発揮して、良いかどうか分からないで、大きい牛が1、2を提げることができます