利用编程从Hbase读取学生成绩，并实现每个学生成绩平均分和每门课程成绩平均分。要求用MapReduce Java API编程

好的，这个问题需要使用 Hadoop MapReduce 框架来实现。首先，需要编写一个 Mapper 类来解析 HBase 表中的每个行数据，并将学生 ID 和其成绩数据发送给 Reducer。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class ScoresMapper extends Mapper<Object, Result, Text, FloatWritable> {

    private final Text studentId = new Text();
    private final FloatWritable score = new FloatWritable();

    public void map(Object key, Result value, Context context) throws IOException, InterruptedException {
        // 解析行数据
        String rowKey = Bytes.toString(value.getRow());
        String[] parts = rowKey.split("\\|");
        String studentId = parts[0];
        String subject = parts[1];
        float score = Bytes.toFloat(value.getValue(Bytes.toBytes("score"), Bytes.toBytes("value")));

        // 将学生 ID 和成绩数据发送给 Reducer
        context.write(new Text(studentId), new FloatWritable(score));
        context.write(new Text(subject), new FloatWritable(score));
    }
}

接下来，需要编写一个 Reducer 类来计算每个学生的平均分和每门课程的平均分。

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class ScoresReducer extends Reducer<Text, FloatWritable, Text, FloatWritable> {

    private final FloatWritable result = new FloatWritable();

    public void reduce(Text key, Iterable<FloatWritable> values, Context context) throws IOException, InterruptedException {
        List<Float> scores = new ArrayList<Float>();

        // 收集所有成绩数据
        for (FloatWritable value : values) {
            scores.add(value.get());
        }

        // 计算平均分
        float totalScore = 0;
        for (Float score : scores) {
            totalScore += score;
        }
        float averageScore = totalScore / scores.size();

        // 发送结果给输出
        result.set(averageScore);
        context.write(key, result);
    }
}

最后，需要在主函数中配置并运行 MapReduce 作业。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

public class ScoresJob {

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();

        // 配置 HBase 表扫描
        Scan scan = new Scan();
        FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        filters.addFilter(new PrefixFilter(Bytes.toBytes("")));
        filters.addFilter(new KeyOnlyFilter());
        filters.addFilter(new FirstKeyOnlyFilter());
        scan.setFilter(filters);

        // 配置 MapReduce 作业
        Job job = Job.getInstance(conf, "Student Scores");
        job.setJarByClass(ScoresJob.class);
        TableMapReduceUtil.initTableMapperJob("student_scores", scan, ScoresMapper.class, Text.class, FloatWritable.class, job);
        job.setReducerClass(ScoresReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FloatWritable.class);

        // 运行作业并等待完成
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

这段代码将 HBase 表扫描和 MapReduce 作业配置在一起，并启动作业运行。注意，在这个例子中，我们使用了 HBase 表扫描来过滤出所有的行数据，因此 Mapper 类不需要实现 InputFormat 接口。