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

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

// MatrixMultiplication.java
import java.io.IOException;
import java.util.StringTokenizer;

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

public class MatrixMultiplication {

  public static class Mapper1
       extends Mapper<Object, Text, Text, Text>{

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

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      Configuration conf = context.getConfiguration();
      int matrixM = Integer.parseInt(conf.get("matrixM"));
      int matrixP = Integer.parseInt(conf.get("matrixP"));

      StringTokenizer tokenizer = new StringTokenizer(value.toString(), ",");
      int i = 0;
      int j = 0;
      String[] rowVector = new String[matrixP];
      while (tokenizer.hasMoreTokens()) {
        if (j < matrixP) {
          rowVector[j] = tokenizer.nextToken();
        } else {
          break;
        }
        j++;
      }

      while (tokenizer.hasMoreTokens()) {
        String valueText = tokenizer.nextToken();
        for (int k = 0; k < matrixM; k++) {
          outKey.set(i + "," + k);
          outValue.set("A," + j + "," + valueText);
          context.write(outKey, outValue);
        }
        i++;
      }

      for (int k = 0; k < matrixP; k++) {
        outKey.set(i + "," + k);
        outValue.set("B," + j + "," + rowVector[k]);
        context.write(outKey, outValue);
      }
    }
  }

  public static class Reducer1
       extends Reducer<Text,Text,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<Text> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int[] vectorA = new int[10];
      int[] vectorB = new int[10];
      for (Text val : values) {
        String[] valueTexts = val.toString().split(",");
        if (valueTexts[0].equals("A")) {
          vectorA[Integer.parseInt(valueTexts[1])] = Integer.parseInt(valueTexts[2]);
        } else {
          vectorB[Integer.parseInt(valueTexts[1])] = Integer.parseInt(valueTexts[2]);
        }
      }

      int dotProduct = 0;
      for (int i = 0; i < vectorA.length; i++) {
        dotProduct += vectorA[i] * vectorB[i];
      }

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

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    conf.set("matrixM", args[2]);
    conf.set("matrixP", args[3]);
    Job job = Job.getInstance(conf, "matrix multiplication");
    job.setJarByClass(MatrixMultiplication.class);
    job.setMapperClass(Mapper1.class);
    job.setReducerClass(Reducer1.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

运行时需要提供四个参数，分别是输入文件路径、输出文件路径、矩阵M的大小和矩阵P的大小。例如：

hadoop jar MatrixMultiplication.jar input output 2 3

这里的输入文件是一个文本文件，格式如下：

1,2,3
4,5,6

表示一个2x3的矩阵。输出文件则是一个文本文件，格式如下：

0,0    14
0,1    32
0,2    50
1,0    32
1,1    77
1,2    122

表示两个矩阵相乘的结果。