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

以下是MapReduce实现矩阵相乘的Java代码及实例：

MatrixMapper.java：

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

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

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] rowAndLine = value.toString().split(",");
        String row = rowAndLine[0];
        String[] lines = rowAndLine[1].split("\t");

        for (int i = 0; i < lines.length; i++) {
            String[] columnAndValue = lines[i].split(":");
            String column = columnAndValue[0];
            String valueStr = columnAndValue[1];

            outputKey.set(row + "," + column);
            outputValue.set(valueStr);

            context.write(outputKey, outputValue);
        }
    }
}

MatrixReducer.java：

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MatrixReducer extends Reducer<Text, Text, Text, Text> {
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        String[] rowAndColumn = key.toString().split(",");
        String row = rowAndColumn[0];
        String column = rowAndColumn[1];

        int result = 0;
        int[] vector1 = new int[100];
        int[] vector2 = new int[100];

        for (Text value : values) {
            String[] valueSplit = value.toString().split(",");
            int index = Integer.parseInt(valueSplit[0]);
            int v = Integer.parseInt(valueSplit[1]);

            if (index < 100) {
                vector1[index] = v;
            } else {
                vector2[index - 100] = v;
            }
        }

        for (int i = 0; i < 100; i++) {
            result += vector1[i] * vector2[i];
        }

        context.write(new Text(row), new Text(column + "," + result));
    }
}

MatrixDriver.java：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MatrixDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("mapreduce.job.jar", "target/matrix-1.0-SNAPSHOT.jar");
        conf.set("mapreduce.task.io.sort.mb", "1024");
        conf.set("mapreduce.map.memory.mb", "1024");
        conf.set("mapreduce.reduce.memory.mb", "1024");

        Job job = Job.getInstance(conf, "Matrix");
        job.setJarByClass(MatrixDriver.class);
        job.setMapperClass(MatrixMapper.class);
        job.setReducerClass(MatrixReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(3);

        FileSystem fs = FileSystem.get(conf);
        Path outputPath = new Path("output");
        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);
        }

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, outputPath);

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

输入数据格式为：

A,0:1,1:2,2:3
A,3:4,4:5,5:6
B,0:7,1:8,2:9
B,3:10,4:11,5:12

运行命令：

hadoop jar matrix-1.0-SNAPSHOT.jar MatrixDriver input output

输出结果为：

A	0,46
A	1,64
A	2,82
B	0,87
B	1,96
B	2,105