码迷，mamicode.com
首页 > 其他好文 > 详细

MapReduce实现二阶矩阵相乘

时间：2016-03-29 10:52:36      阅读：178      评论：0      收藏：0      [点我收藏+]

标签：

二阶矩阵相乘公式

技术分享

上例中的C11=A11*B11+A12*B21+A13*B31=1*3+0*2+2*1=5、C12=A11*B12+A12*B22+A13*B32=1*1+0*1+2*0=1


分析

 因为分布式计算的特点，需要找到相互独立的计算过程，以便能够在不同的节点上进行计算而不会彼此影响。根据矩

阵乘法的公式，C中各个元素的计算都是相互独立的，即各个cij在计算过程中彼此不影响。这样的话，在Map阶段可

以把计算所需要的元素都集中到同一个key中，然后，在Reduce阶段就可以从中解析出各个元素来计算cij。  另外，

以a11为例，它将会在c11、c12...c1p的计算中使用，以b11为例，它将会在c11、c21...cm1的计算中使用，也就是说，在Map阶段，当我们从HDFS取出一行记录时，如

果该记录是A的元素，则需要存储成p个<key, value>对，并且这p个key互不相同；如果该记录是B的元素，则需要存

储成m个<key, value>对，同样的，m个key也应互不相同；但同时，用于存放计算cij的ai1、ai2……ain和b1j、

b2j……bnj的<key, value>对的key应该都是相同的，这样才能被传递到同一个Reduce中。


设计

普遍有一个共识是：数据结构+算法=程序，所以在编写代码之前需要先理清数据存储结构和处理数据的算法。

Map阶段

在Map阶段，需要做的是进行数据准备。把来自矩阵A的元素aij，标识成p条<key, value>的形式，key="i,k",（其中

k=1,2,...,p），value="A,j,Aij"；把来自矩阵B的元素bij，标识成m条<key, value>形式，key="k,j"（其中

k=1,2,...,m），value="B,i,Bij"。  经过处理，用于计算cij需要的a、b就转变为有相同key（"i,j"）的数据对，通过value

中"A"、"B"能区分元素是来自矩阵A还是矩阵B，以及具体的位置（在矩阵A的第几列，在矩阵B的第几行）。

Shuffle阶段

这个阶段是Hadoop自动完成的阶段，具有相同key的value被分到同一个list中，形成<key,list(value)>对，再传递给Reduce。

Reduce阶段 

在Reduce阶段，有两个问题需要解决：

a. 当前的<key, list(value)>对是为了计算矩阵C的哪个元素？因为map阶段对数据的处理，key（i,j）中的数据对，就

是其在矩阵C中的位置，第i行j列。

b. list中的每个value是来自矩阵A或矩阵B的哪个位置？这个也在map阶段进行了标记，对于value（x,y,z），只需要找

到y相同的来自不同矩阵（即x分别为A和B）的两个元素，取z相乘，然后加和即可。

矩阵的两种表示方式

矩阵常用的两种表示方式，第一种是原始的表示方式，第二种是稀疏矩阵（只存储非0的元素）的表示方式。

第一种：使用最原始的表示方式，相同行内不同列数据通过","分割，不同行通过换行分割； 

第二种：通过行列表示法，即文件中的每行数据有三个元素通过分隔符分割，第一个元素表示行，第二个元素表示

列，第三个元素表示数据。这种方式对于可以不列出为0的元素，即可以减少稀疏矩阵的数据量。 

技术分享



编写代码：

第一种数据结构

查看源数据：

[hadoop@master ~]$ hadoop fs -cat /user/hdfs/matrix/B
10,15
0,2
11,9
[hadoop@master ~]$ hadoop fs -cat /user/hdfs/matrix/A
1,2,3
4,5,0
7,8,9
10,11,12

MartrixMultiply：

package com.oner.mr.matrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.oner.mr.util.HdfsDAO;

public class MartrixMultiply {

	public static class MatrixMapper extends
			Mapper<LongWritable, Text, Text, Text> {

		private String flag;// A or B;
		private int m = 4;// 矩阵A的行数
		private int p = 2;// 矩阵B的列数
		private int rowIndexA = 1; // 矩阵A，当前在第几行
		private int rowIndexB = 1; // 矩阵B，当前在第几行

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			FileSplit split = (FileSplit) context.getInputSplit();
			flag = split.getPath().getName();// // 得到读取的矩阵名称
		}

		@Override
		protected void map(LongWritable k1, Text v1, Context context)
				throws IOException, InterruptedException {
			// 切分每行数据
			String[] fields = MainRun.DELIMITER.split(v1.toString());

			if (flag.equals("A")) {// 如果读的是矩阵A，则fields格式为{1,2,3},数组长度为3
				for (int k = 1; k <= p; k++) {// p表示矩阵B的列数
					Text key = new Text(rowIndexA + "," + k);//
					for (int j = 0; j < fields.length; j++) {// j代表矩阵A的当前列,fields.length表示矩阵A的列数，等于矩阵B的行数
						Text value = new Text("A," + (j + 1) + "," + fields[j]);// v的值为
						context.write(key, value);// 输出的数据格式key为(i,k),value为(A,j,Aij)。
						System.out.println(key.toString() + "  "
								+ value.toString());
					}

				}
				rowIndexA++; // 每执行一次map方法，矩阵向下移动一行

			} else if (flag.equals("B")) {// 如果读的是B,fields的格式为{10,15},数组长度为2
				for (int k = 1; k <= m; k++) {// m表示矩阵A的行数
					for (int j = 0; j < fields.length; j++) {// fields.length表示矩阵B的列数
						Text key = new Text(k + "," + (j + 1));
						Text value = new Text("B:" + rowIndexB + ","
								+ fields[j]);
						context.write(key, value);// 输出的数据格式key为(k,j),value为(B,i,Bij)。
						System.out.println(key.toString() + "  "
								+ value.toString());
					}
				}
				rowIndexB++;// 每执行一次map方法，矩阵向下移动一行
			}
		}

	}

	public static class MatrixReducer extends
			Reducer<Text, Text, Text, LongWritable> {
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Map<String, String> mapA = new HashMap<String, String>();
			Map<String, String> mapB = new HashMap<String, String>();

			System.out.print(key.toString() + ":");

			for (Text value : values) {
				String val = value.toString();
				System.out.print("(" + val + ")");

				if (val.startsWith("A")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到A,j,Aij中的j,Aij
					mapA.put(kv[0], kv[1]);// 将j作为key,Aij作为value存入mapA

					// System.out.println("A:" + kv[0] + "," + kv[1]);

				} else if (val.startsWith("B")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到B,j,Bij中的i,Bij
					mapB.put(kv[0], kv[1]);// 将i作为key,Bij作为value存入mapB

					// System.out.println("B:" + kv[0] + "," + kv[1]);
				}
			}

			long result = 0;
			Iterator<String> mkeys = mapA.keySet().iterator();// 得到mapA所有的键集合
			while (mkeys.hasNext()) {
				String mkey = mkeys.next();
				if (mapB.get(mkey) == null) {// 因为mkey取的是mapA的key集合，所以只需要判断mapB是否存在即可。
					continue;
				}
				result += Long.parseLong(mapA.get(mkey))
						* Long.parseLong(mapB.get(mkey));
			}
			context.write(key, new LongWritable(result));
			System.out.println();

			// System.out.println("C:" + key.toString() + "," + result);

		}
	}

	public static void run(Map<String, String> path) throws IOException,
			ClassNotFoundException, InterruptedException {

		Configuration conf = new Configuration();

		String input = path.get("input");
		String input1 = path.get("input1");
		String input2 = path.get("input2");
		String output = path.get("output");

		HdfsDAO hdfs = new HdfsDAO(MainRun.HDFS, conf);
		hdfs.rmr(input);
		hdfs.mkdirs(input);
		hdfs.copyFile(path.get("A"), input1);
		hdfs.copyFile(path.get("B"), input2);

		Job job = Job.getInstance(conf);
		job.setJarByClass(MainRun.class);

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

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

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

		FileInputFormat.setInputPaths(job, new Path(input1), new Path(input2));// 加载2个输入数据集
		FileOutputFormat.setOutputPath(job, new Path(output));

		job.waitForCompletion(true);
	}
}

MainRun：

package com.oner.mr.matrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

/*
 * 驱动程序
 */
public class MainRun {

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

	public static void main(String[] args) throws ClassNotFoundException,
			IOException, InterruptedException {
		martrixMultiply();
	}

	private static void martrixMultiply() throws ClassNotFoundException,
			IOException, InterruptedException {
		Map<String, String> path = new HashMap<String, String>();
		path.put("A", "/home/hadoop/logfile/matrix/A.csv");// 本地的数据文件
		path.put("B", "/home/hadoop/logfile/matrix/B.csv");
		path.put("input", HDFS + "/user/hdfs/matrix");// HDFS的目录
		path.put("input1", HDFS + "/user/hdfs/matrix/A");
		path.put("input2", HDFS + "/user/hdfs/matrix/B");
		path.put("output", HDFS + "/user/hdfs/matrix/output");

		MartrixMultiply.run(path);
	}

}

打成jar包后运行：hadoop jar matrix.jar com.oner.mr.matrix.MainRun

查看结果：

[hadoop@master ~]$ hadoop fs -cat /user/hdfs/matrix/output/part-r-00000
1,1	43
1,2	46
2,1	40
2,2	70
3,1	169
3,2	202
4,1	232
4,2	280
绘图演示结果：

技术分享


第二种数据结构

MainRun：

package com.oner.mr.sparsematrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

/*
 * 驱动程序
 */
public class MainRun {

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

	public static void main(String[] args) throws ClassNotFoundException,
			IOException, InterruptedException {
		sparseMartrixMultiply();
	}

	private static void sparseMartrixMultiply() throws ClassNotFoundException,
			IOException, InterruptedException {
		Map<String, String> path = new HashMap<String, String>();
		path.put("A", "/home/hadoop/logfile/matrix2/A.csv");// 本地的数据文件
		path.put("B", "/home/hadoop/logfile/matrix2/B.csv");
		path.put("input", HDFS + "/user/hdfs/matrix2");// HDFS的目录
		path.put("input1", HDFS + "/user/hdfs/matrix2/A");
		path.put("input2", HDFS + "/user/hdfs/matrix2/B");
		path.put("output", HDFS + "/user/hdfs/matrix2/output");

		MartrixMultiply.run(path);
	}

}

MartrixMultiply：

package com.oner.mr.sparsematrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.oner.mr.util.HdfsDAO;

public class MartrixMultiply {

	public static class SparseMatrixMapper extends
			Mapper<LongWritable, Text, Text, Text> {

		private String flag;// A or B;
		private int m = 4;// 矩阵A的行数
		private int p = 2;// 矩阵B的列数

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			FileSplit split = (FileSplit) context.getInputSplit();
			flag = split.getPath().getName();// // 得到读取的矩阵名称
		}

		@Override
		protected void map(LongWritable k1, Text v1, Context context)
				throws IOException, InterruptedException {
			// 切分每行数据
			String[] fields = MainRun.DELIMITER.split(v1.toString());

			if ("A".equals(flag)) {
				for (int i = 1; i <= p; i++) {
					context.write(new Text(fields[0] + "," + i), new Text("A,"
							+ fields[1] + "," + fields[2]));
				}
			} else if ("B".equals(flag)) {
				for (int i = 1; i <= m; i++) {
					context.write(new Text(i + "," + fields[1]), new Text("B,"
							+ fields[0] + "," + fields[2]));
				}
			}

		}

	}

	public static class SparseMatrixReducer extends
			Reducer<Text, Text, Text, LongWritable> {
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Map<String, String> mapA = new HashMap<String, String>();
			Map<String, String> mapB = new HashMap<String, String>();

			for (Text value : values) {
				String val = value.toString();

				if (val.startsWith("A")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到A,j,Aij中的j,Aij
					mapA.put(kv[0], kv[1]);// 将j作为key,Aij作为value存入mapA

				} else if (val.startsWith("B")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到B,j,Bij中的i,Bij
					mapB.put(kv[0], kv[1]);// 将i作为key,Bij作为value存入mapB

				}
			}

			long result = 0;
			// 可能在mapA中存在在mapB中不存在的key，或相反情况
			// 因为，数据定义的时候使用的是稀疏矩阵的定义
			// 所以，这种只存在于一个map中的key，说明其对应元素为0，不影响结果
			Iterator<String> mkeys = mapA.keySet().iterator();// 得到mapA所有的键集合
			while (mkeys.hasNext()) {
				String mkey = mkeys.next();
				if (mapB.get(mkey) == null) {// 因为mkey取的是mapA的key集合，所以只需要判断mapB是否存在即可。
					continue;
				}
				result += Long.parseLong(mapA.get(mkey))
						* Long.parseLong(mapB.get(mkey));
			}
			context.write(key, new LongWritable(result));

		}
	}

	public static void run(Map<String, String> path) throws IOException,
			ClassNotFoundException, InterruptedException {

		Configuration conf = new Configuration();

		String input = path.get("input");
		String input1 = path.get("input1");
		String input2 = path.get("input2");
		String output = path.get("output");

		HdfsDAO hdfs = new HdfsDAO(MainRun.HDFS, conf);
		hdfs.rmr(input);
		hdfs.mkdirs(input);
		hdfs.copyFile(path.get("A"), input1);
		hdfs.copyFile(path.get("B"), input2);

		Job job = Job.getInstance(conf);
		job.setJarByClass(MainRun.class);

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

		job.setMapperClass(SparseMatrixMapper.class);
		job.setReducerClass(SparseMatrixReducer.class);

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

		FileInputFormat.setInputPaths(job, new Path(input1), new Path(input2));// 加载2个输入数据集
		FileOutputFormat.setOutputPath(job, new Path(output));

		job.waitForCompletion(true);
	}
}

打成jar包运行：hadoop jar matrix

查看结果：

[hadoop@master matrix2]$ hadoop fs -cat /user/hdfs/matrix2/output/part-r-00000
1,1	43
1,2	46
2,1	40
2,2	70
3,1	169
3,2	202
4,1	232
4,2	280


MapReduce实现二阶矩阵相乘

标签：

原文地址：http://blog.csdn.net/xw_classmate/article/details/50951545
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！