MapReduce实现矩阵相乘java代码及实例

以下是一个简单的 Java 实现 MapReduce 矩阵相乘的例子：

首先是 Mapper 类：

public class MatrixMapper extends Mapper<LongWritable, Text, Text, Text> {
    private Text outputKey = new Text();
    private Text outputValue = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] line = value.toString().split(",");
        String row = line[0];
        String col = line[1];
        String val = line[2];

        if (context.getMapperNumTasks() == 2) {
            // Matrix A
            for (int i = 0; i < context.getNumReduceTasks(); i++) {
                outputKey.set(row + "," + i);
                outputValue.set("A," + col + "," + val);
                context.write(outputKey, outputValue);
            }
        } else {
            // Matrix B
            for (int i = 0; i < context.getNumReduceTasks(); i++) {
                outputKey.set(i + "," + col);
                outputValue.set("B," + row + "," + val);
                context.write(outputKey, outputValue);
            }
        }
    }
}

这个 Mapper 会读入一个 CSV 文件，CSV 文件中的每一行都表示一个矩阵元素，包括行、列、和值。Mapper 会将每个元素拆分为三个部分，并将它们输出为键值对，其中键由行和列组成，值由矩阵标识符（A 或 B）、列或行编号以及值组成。这些键值对将作为 Reduce 阶段的输入。

接下来是 Reducer 类：

public class MatrixReducer extends Reducer<Text, Text, Text, DoubleWritable> {
    private DoubleWritable outputValue = new DoubleWritable();

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Map<Integer, Double> valuesA = new HashMap<>();
        Map<Integer, Double> valuesB = new HashMap<>();

        for (Text value : values) {
            String[] parts = value.toString().split(",");
            String matrix = parts[0];
            int index = Integer.parseInt(parts[1]);
            double val = Double.parseDouble(parts[2]);

            if (matrix.equals("A")) {
                valuesA.put(index, val);
            } else {
                valuesB.put(index, val);
            }
        }

        double result = 0.0;
        for (int i = 0; i < context.getNumReduceTasks(); i++) {
            result += valuesA.getOrDefault(i, 0.0) * valuesB.getOrDefault(i, 0.0);
        }

        outputValue.set(result);
        context.write(key, outputValue);
    }
}

这个 Reducer 会读入上述 Mapper 输出的键值对。Reducer 会将键拆分为行和列编号，并且将值分为来自矩阵 A 和矩阵 B 的两部分。Reducer 会将两个部分分别存储在两个 Map 中，并且使用这些值计算矩阵乘积的一个元素。最后，Reducer 会将计算结果输出为键值对。

最后是主程序：

public class MatrixMultiplication {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Matrix Multiplication");

        job.setJarByClass(MatrixMultiplication.class);
        job.setMapperClass(MatrixMapper.class);
        job.setReducerClass(MatrixReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.setInputPaths(job, new Path(args[0]));
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setNumReduceTasks(Integer.parseInt(args[2]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

这个程序会读入三个参数：输入文件路径、输出文件路径和 Reduce 数量。它会创建一个 MapReduce 作业，并且将 Mapper、Reducer、输入格式、输出格式和输入输出路径设置为作业的属性。最后，它会运行作业并等待其完成。

这个程序可以使用以下命令来运行：

hadoop jar matrix-multiplication.jar input.csv output 2

其中，`input.csv` 是输入文件，`output` 是输出目录，`2` 是 Reduce 数量。