使用 MapReduce 实现分组排名

题目：

1.输出男女组各前3名。

2.分组降序输出所有人的成绩。

3.对应的SQL语句。

数据：

scores: name,age,gender,score

解答如下：

1. 输出男女组各前3名。

Java代码

class ContestStatTop3Mapper extends Mapper<Object, Text, Text, Text> {

  private Text outKey = new Text();
  private Text outValue = new Text();

  @Override
  public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    // name, age, gender, score
    String[] arr = value.toString().split(",");
    outKey.set(arr[2]); // gender
    outValue.set(arr[0] + "," + arr[1] + "," + arr[3]); // name,age,score
    context.write(outKey, outValue);
  }
}

class ContestStatTop3Reducer extends Reducer<Text, Text, Text, Text> {


  private Text outKey = new Text();
  private Text outValue = new Text();

  private int maxSize = 3;

  private List<ContestPerson> list = new ArrayList<>(maxSize);

  @Override
  public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    String gender = key.toString();
    int pos =0;
    int minScore = Integer.MAX_VALUE;
    list.clear();
    for (Text t : values) {
      String[] arr = t.toString().split(",");
      // name,age,score
      String name = arr[0];
      int age = Integer.parseInt(arr[1]);
      int score = Integer.parseInt(arr[2]);
      ContestPerson person = new ContestPerson(name,age,gender,score);
      if (list.size() < maxSize) { //always add if the size is less than maxSize
        list.add(person);
        if (score <= minScore) { // update minScore
          minScore = score;
          pos = list.size() - 1;// remember the position of person with minScore
        }
      } else {
        if(score > minScore){
        // remove the person with minScore
          list.remove(pos);
        // add the new person
          list.add(person);
        // update minScore
          minScore = person.score;
          pos = list.size() -1;
          for(int i = 0; i< list.size();i++){
            if(minScore > list.get(i).score){
              minScore = list.get(i).score;
              pos =i;
            }
          }
        }
      }
    }
    //output
    outKey.set(gender);
    for(ContestPerson person: list){
      outValue.set(person.toString());
      context.write(outKey, outValue);

    }
  }
}

SQL 代码

# option 1
select m.name,m.age,m.gender,m.score from scores as m
where 3>= (
select count(*) from scores as s where m.gender = s.gender and m.score <= s.score
)
order by m.gender,m.score desc,m.age,m.name;

# option 2
select m.name,m.age,m.gender,m.score from scores as m
left join scores as s on m.gender = s.gender and m.score <=  s.score
group by m.name,m.age,m.gender,m.score
having count(*) <=3
order by m.gender,m.score desc,m.age,m.name;

2.分组降序输出所有人的成绩。

Java代码

class ContestStatSortMapper extends Mapper<Object, Text, Text, Text> {

  private Text outKey = new Text();
  private Text outValue = new Text();

  @Override
  public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    // name, age, gender, score
    String[] arr = value.toString().split(",");
    outKey.set(arr[2]); // gender
    outValue.set(arr[0] + "," + arr[1] + "," + arr[3]); // name,age,score
    context.write(outKey, outValue);
  }
}

class ContestStatSortReducer extends Reducer<Text, Text, Text, Text> {


  private Text outKey = new Text();
  private Text outValue = new Text();

  private int maxSize = 3;

  private List<ContestPerson> list = new ArrayList<>(maxSize);

  @Override
  public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    String gender = key.toString();
    list.clear();
    for (Text t : values) {
      String[] arr = t.toString().split(",");
      // name,age,score
      String name = arr[0];
      int age = Integer.parseInt(arr[1]);
      int score = Integer.parseInt(arr[2]);
      ContestPerson person = new ContestPerson(name,age,gender,score);
      list.add(person);
    }
    //sort by score desc, then by age asc, then by name asc
    Collections.sort(list, new Comparator<ContestPerson>() {
      @Override
      public int compare(ContestPerson o1, ContestPerson o2) {
        if (o1.score > o2.score) {
          return -1;
        } else if (o1.score < o2.score) {
          return 1;
        } else {
          if (o1.age > o2.age) {
            return 1;
          } else if (o1.age < o2.age) {
            return -1;
          } else {
            return o1.name.compareTo(o2.name);
          }
        }
      }
    });
    //output
    outKey.set(gender);
    for(ContestPerson person: list){
      outValue.set(person.toString());
      context.write(outKey, outValue);
    }
  }
}

SQL代码

select m.name,m.age,m.gender,m.score from scores as m
order by m.gender,m.score desc,m.age,m.name; 

附：Java代码中所使用的ContestPerson类

class ContestPerson extends  Object{
  public String name;
  public int age;
  public String gender;
  public int score;
  public ContestPerson(String name, int age, String gender,int score){
    this.name = name;
    this.age = age;
    this.gender = gender;
    this.score = score;
  }
  public String toString(){
    return this.name + ","+ this.age + "," + this.gender + "," + this.score;
  }
}