Hadoop MapReduce实现矩阵的乘法

算法的实现思路：

图片和思路来自于http://blog.fens.me/hadoop-mapreduce-matrix/

• 新建2个矩阵数据文件：m1, m2

• m1

  
  1,0,2
  -1,3,1
  

  m2

  
  3,1
  2,1
  1,0

  
  
• 新建启动程序:MainRun.java
• 新建MR程序:MartrixMultiply.java

MainRun.java

public class MainRun {

    public static final String HDFS = "hdfs://10.103.240.160:9000";
    public static final Pattern DELIMITER = Pattern.compile("[\t,]");

    public static void main(String[] args) {
        martrixMultiply();
    }
    
    public static void martrixMultiply() {
        Map<String, String> path = new HashMap<String, String>();
        path.put("m1", "logfile/matrix/m1.csv");// 本地的数据文件
        path.put("m2", "logfile/matrix/m2.csv");
        path.put("input", HDFS + "/usr/hadoop/matrix");// HDFS的目录
        path.put("input1", HDFS + "/usr/usr/matrix/m1");
        path.put("input2", HDFS + "/usr/usr/matrix/m2");
        path.put("output", HDFS + "/usr/hadoop/matrix/output");

        try {
        	MartrixMultiply.run(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.exit(0);
    }
}

MartrixMultiply.java

public class MartrixMultiply {

	public static class MyMapper extends Mapper<LongWritable, Text, Text, Text> {
		private String flag; // m1 or m2
		private int rowNumA = 2; // 矩阵A的行数，因为要在对B的矩阵处理中要用
		private int colNumA = 3;// 矩阵A的列数
		private int rolNumB = 3;
		private int colNumB = 2;// 矩阵B的列数
		private int rowIndexA = 1; // 矩阵A，当前在第几行
		private int rowIndexB = 1; // 矩阵B，当前在第几行
		private static final Text k = new Text();
		private static final Text v = new Text();

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			FileSplit split = (FileSplit) context.getInputSplit();
			flag = split.getPath().getName();// 判断读的数据集
		}

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] elements = value.toString().split(",");
			/*
			 * i表示在这一行中，该元素是第一个元素 j表示该行应与B矩阵的哪一列进行相乘 k.set(rowIndexA + "," + (j
			 * + 1)); 行:是由rowIndexA决定的 列：是由该行与矩阵B相乘的那一列决定的
			 */
			if (flag.equals("m1")) {
				// 对于每一个elements元素进行处理，elements.length=colNumA
				for (int i = 0; i < colNumA; i++) {
					for (int j = 0; j < colNumB; j++) {
						k.set(rowIndexA + "," + (j + 1));
						v.set("A" + ":" + (i + 1) + "," + elements[i]);
						context.write(k, v);
					}
				}
				rowIndexA++;

			}
			/*
			 * i表示在这一行中，该元素是第几个元素 j表示A矩阵的第几行与该矩阵进行相乘 k.set((j+1) + "," +
			 * (i+1))表示该元素的在C矩阵中位置 行：是由与该元素相乘的A矩阵的第几行决定的 列:是有该元素在B矩阵中第几列决定的
			 * rowIndexB决定了该元素在相乘得到的和中占第几个位置
			 */
			else if (flag.equals("m2")) {
				for (int i = 0; i < colNumB; i++) {
					for (int j = 0; j < rowNumA; j++) {
						k.set((j + 1) + "," + (i + 1));
						v.set("B:" + rowIndexB + "," + elements[i]);
						context.write(k, v);
					}
				}
			}
			rowIndexB++;
		}
	}

	public static class MyReducer extends
			Reducer<Text, Text, Text, IntWritable> {
		private static int[] a = new int[3];
		private static int[] b = new int[3];
		private static IntWritable v = new IntWritable();

		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			for (Text value : values) {
				String[] vs = value.toString().split(":");
				if (vs[0].equals("A")) {
					String[] ints = vs[1].toString().split(",");
					a[Integer.parseInt(ints[0])-1] = Integer.parseInt(ints[1]);
				} else {
					String[] ints = vs[1].toString().split(",");
					b[Integer.parseInt(ints[0])-1] = Integer.parseInt(ints[1]);
				}
			}
				v.set(a[0] * b[0] + a[1] * b[1] + a[2] * b[2]);
				context.write(key, v);
		}

	}

	public static void run(Map<String, String> path) throws Exception {
		String input = path.get("input");
		String input1 = path.get("input1");
		String input2 = path.get("input2");
		String output = path.get("output");

		Configuration conf = new Configuration();
		final FileSystem fileSystem = FileSystem.get(new URI(input), conf);
		final Path outPath = new Path(output);
		if (fileSystem.exists(outPath)) {
			fileSystem.delete(outPath, true);
		}
		conf.set("hadoop.job.user", "hadoop");
//		conf.set("mapred.job.tracker", "10.103.240.160:9001");

		final Job job = new Job(conf);
		FileInputFormat.setInputPaths(job, input);
		job.setJarByClass(MartrixMultiply.class);
		job.setMapperClass(MyMapper.class);
		job.setReducerClass(MyReducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		job.setNumReduceTasks(1);// 设置个数为1
		FileOutputFormat.setOutputPath(job, outPath);
		job.waitForCompletion(true);
	}

}