mapreduce项目统计hbase成绩表中 单科最高单科排名、总分排名代码

### 回答1：
首先，在 HBase 中，我们需要创建一个成绩表，表中有如下几列：

- 学号（row key）：用于标识每一个学生的唯一标识
- 姓名：学生的姓名
- 科目：学生所修的科目名称（如：数学、英语、物理等）
- 成绩：学生在某一科目上的成绩

假设我们已经在 HBase 中创建了一个名为 "scores" 的成绩表，现在我们要使用 MapReduce 统计出单科最高单科排名和总分排名。

在 MapReduce 中，我们需要编写两个类：一个是 Mapper 类，一个是 Reducer 类。

首先是 Mapper 类，我们可以让它从 HBase 表中读取每一个学生的信息，并且输出两组键值对：一组是单科成绩最高的科目和成绩，另一组是总分和成绩。

public static class ScoreMapper extends TableMapper<Text, DoubleWritable> {
  private Text subject = new Text();
  private DoubleWritable score = new DoubleWritable();

  @Override
  public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
    // 读取学生的科目和成绩
    String subjectName = Bytes.toString(value.getValue(Bytes.toBytes("info"), Bytes.toBytes("subject")));
    double subjectScore = Bytes.toDouble(value.getValue(Bytes.toBytes("info"), Bytes.toBytes("score")));

    // 输出单科成绩最高的科目和成绩
    subject.set(subjectName);
    score.set(subjectScore);
    context.write(subject, score);

    // 输出总分  

### 回答2：
在MapReduce项目中，统计HBase成绩表中单科最高单科排名和总分排名的代码如下：

// 定义Mapper类
public class ScoreMapper extends TableMapper<Text, IntWritable> {
private Text subject = new Text();
private IntWritable score = new IntWritable(0);

public void map(ImmutableBytesWritable rowKey, Result result, Context context)
throws IOException, InterruptedException {
// 获取成绩表中的列族和列名
String columnFamily = "cf";
String subjectColumn = "subject";
String scoreColumn = "score";

// 读取学科和成绩
byte[] subjectBytes = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(subjectColumn));
byte[] scoreBytes = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(scoreColumn));

// 转换成String类型，并设置输出键值对
subject.set(Bytes.toString(subjectBytes));
score.set(Bytes.toInt(scoreBytes));
context.write(subject, score);
}
}

// 定义Reducer类
public class ScoreReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text subject, Iterable<IntWritable> scores, Context context)
throws IOException, InterruptedException {
int maxScore = Integer.MIN_VALUE;
for (IntWritable score : scores) {
maxScore = Math.max(maxScore, score.get());
}
// 输出单科最高分
context.write(subject, new IntWritable(maxScore));
}
}

// 定义新的Mapper类，用于计算总分
public class TotalScoreMapper extends TableMapper<Text, IntWritable> {
private Text student = new Text();
private IntWritable score = new IntWritable(0);

public void map(ImmutableBytesWritable rowKey, Result result, Context context)
throws IOException, InterruptedException {
// 获取学生姓名和总分
byte[] studentBytes = result.getRow();
byte[] scoreBytes = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(scoreColumn));

// 转换成String类型，并设置输出键值对
student.set(Bytes.toString(studentBytes));
score.set(Bytes.toInt(scoreBytes));
context.write(student, score);
}
}

// 定义新的Reducer类，用于计算总分排名
public class TotalScoreReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text student, Iterable<IntWritable> scores, Context context)
throws IOException, InterruptedException {
int totalScore = 0;
for (IntWritable score : scores) {
totalScore += score.get();
}
// 输出总分排名
context.write(student, new IntWritable(totalScore));
}
}

// 主函数
public class Main {
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
conf.set(TableInputFormat.INPUT_TABLE, "scores"); // 设置成绩表名

// 计算单科最高单科排名
Job job1 = Job.getInstance(conf, "SubjectMaxScore");
job1.setJarByClass(Main.class);
job1.setInputFormatClass(TableInputFormat.class);
job1.setMapperClass(ScoreMapper.class);
job1.setReducerClass(ScoreReducer.class);
job1.setOutputKeyClass(Text.class);
job1.setOutputValueClass(IntWritable.class);
FileOutputFormat.setOutputPath(job1, new Path("output/subject_max_score"));
job1.waitForCompletion(true);

// 计算总分排名
Job job2 = Job.getInstance(conf, "TotalScoreRank");
job2.setJarByClass(Main.class);
job2.setInputFormatClass(TableInputFormat.class);
job2.setMapperClass(TotalScoreMapper.class);
job2.setReducerClass(TotalScoreReducer.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(IntWritable.class);
FileOutputFormat.setOutputPath(job2, new Path("output/total_score_rank"));
job2.waitForCompletion(true);
}
}

以上代码使用HBase作为输入源，通过两个MapReduce任务分别计算了单科最高分和总分，并将结果输出到指定位置。其中，`TableInputFormat`用于读取HBase中的表数据，`TableMapper`用于处理每一行数据，`TableReducer`用于合并并输出结果。

希望本回答对你有所帮助！  

### 回答3：
在MapReduce项目中，我们可以使用HBase成绩表来统计单科最高单科排名和总分排名。下面是一个可以实现这个功能的简单示例代码：

首先，我们需要定义一个Mapper类来从HBase中读取成绩表的数据，将每个学生的信息拆分为学生ID、科目和分数，然后将科目作为key，学生ID和分数作为value输出。为了保持数据的有序性，我们可以将学生ID加上上升顺序的数字作为中间key。

public class ScoreMapper extends TableMapper<Text, Text> {
private Text outputKey = new Text();
private Text outputValue = new Text();

@Override
protected void map(ImmutableBytesWritable key, Result value, Context context) {
String studentId = Bytes.toString(key.get());
for (Cell cell : value.listCells()) {
String subject = Bytes.toString(CellUtil.cloneQualifier(cell));
String score = Bytes.toString(CellUtil.cloneValue(cell));
outputKey.set(subject);
outputValue.set(studentId + "\t" + score);
context.write(outputKey, outputValue);
}
}

}

然后，我们需要定义一个Reducer类来处理Mapper输出的数据，对每个科目进行排序，并根据分数计算出最高单科排名和总分排名。

public class ScoreReducer extends Reducer<Text, Text, Text, Text> {
private TreeMap<Integer, String> highestSubjectRanking = new TreeMap<>();
private TreeMap<Integer, String> totalRanking = new TreeMap<>();

@Override
protected void reduce(Text key, Iterable<Text> values, Context context) {
TreeMap<Integer, String> ranking = new TreeMap<>();
for (Text val : values) {
String[] parts = val.toString().split("\t");
String studentId = parts[0];
int score = Integer.parseInt(parts[1]);
ranking.put(score, studentId);
}
highestSubjectRanking.put(ranking.lastKey(), ranking.get(ranking.lastKey()));
totalRanking.put(ranking.values().stream().mapToInt(Integer::parseInt).sum(), ranking.get(ranking.lastKey()));
}

@Override
protected void cleanup(Context context) {
int rank = 1;
for (String studentId : highestSubjectRanking.values()) {
context.write(new Text("最高单科排名 第" + rank + "名"), new Text(studentId));
rank++;
}
rank = 1;
for (String studentId : totalRanking.values()) {
context.write(new Text("总分排名 第" + rank + "名"), new Text(studentId));
rank++;
}
}
}

最后，我们需要编写main函数来配置和运行MapReduce任务。

public class ScoreRanking {
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
Job job = Job.getInstance(conf, "Score Ranking");
job.setJarByClass(ScoreRanking.class);
Scan scan = new Scan();
TableMapReduceUtil.initTableMapperJob(
"scores", // HBase成绩表名
scan,
ScoreMapper.class,
Text.class,
Text.class,
job
);
job.setReducerClass(ScoreReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileOutputFormat.setOutputPath(job, new Path(args[0])); // 结果输出路径
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

通过运行以上代码，我们可以将HBase成绩表中的数据按科目进行排序，并输出最高单科排名和总分排名结果。