MapReduce實(shí)現(xiàn)大矩陣乘法(附源碼及詳細(xì)過(guò)程)

《MapReduce實(shí)現(xiàn)大矩陣乘法(附源碼及詳細(xì)過(guò)程)》由會(huì)員分享，可在線閱讀，更多相關(guān)《MapReduce實(shí)現(xiàn)大矩陣乘法(附源碼及詳細(xì)過(guò)程)（15頁(yè)珍藏版）》請(qǐng)?jiān)谘b配圖網(wǎng)上搜索。

1、精選優(yōu)質(zhì)文檔-----傾情為你奉上 報(bào)告題目： MapReduce實(shí)現(xiàn)大矩陣乘法 1、題目要求 題目介紹：請(qǐng)使用MapReduce編程模型實(shí)現(xiàn)矩陣相乘。要求如下： 1.?按照MapReduce框架，對(duì)該問(wèn)題進(jìn)行剖析，并給出解決思路，要求圖文并茂。 2.?展示運(yùn)行代碼并要求注釋清晰。 3.?展示算法運(yùn)行性能（不需要打印最終的矩陣輸出結(jié)果圖），要求分別給出在10^3,?10^4,?10^5維度的矩陣乘法運(yùn)行時(shí)間，map?worker和reduce?worker的實(shí)時(shí)運(yùn)行數(shù)量等。 2、系統(tǒng)環(huán)境 操作系統(tǒng)環(huán)境：Linux Centos6.5 Hadoop環(huán)境：ha

2、doop 2.6.5 完全分布式系統(tǒng)（1個(gè)名稱(chēng)節(jié)點(diǎn)，1個(gè)數(shù)據(jù)節(jié)點(diǎn)） 3、解決方案 3.1 算法分析 假設(shè)有兩個(gè)矩陣A和矩陣B，如下圖所示： 矩陣乘法的數(shù)學(xué)公式如下： 下圖是兩個(gè)矩陣A和B: A，B在文件中的存儲(chǔ)可以是每一行元素占文件的一行，如下圖： 也可以是列表的形式，如下圖： 考慮到大矩陣的稀疏且數(shù)據(jù)量大，HDFS中的文件大小有限制，無(wú)法保證一個(gè)矩陣只保存在一個(gè)文件中，因此采用列表的方式來(lái)存儲(chǔ)矩陣，其格式為。 在Map階段，Mapper按行讀取矩陣A，B所在文件的數(shù)據(jù)，并將其轉(zhuǎn)化為某種格式的鍵值對(duì)保存到上下文。 在Shffle階

3、段，Hadoop自動(dòng)將相同key的value合并成的格式，再將該格式的數(shù)據(jù)傳遞給reduce。 在Reduce階段，Reducer接收Shuffle傳遞過(guò)來(lái)的，處理數(shù)據(jù)得到最終的結(jié)果，并將結(jié)果輸出到目標(biāo)矩陣文件中。 這里的問(wèn)題關(guān)鍵在于，如何設(shè)計(jì)Mapper—Shffle—Reducer之間傳遞的鍵值對(duì)數(shù)據(jù)。 考慮從目標(biāo)矩陣出發(fā)，將Reducer處理的鍵值對(duì)的中的key設(shè)置為目標(biāo)矩陣的行坐標(biāo)和列坐標(biāo)，即”i, j”；那么Iterable(values

4、)迭代器中的數(shù)據(jù)就應(yīng)該是得到Cij需要的數(shù)據(jù)，即 得到這些數(shù)據(jù)還需要標(biāo)注其來(lái)自那個(gè)矩陣，另外根據(jù)公式： 相乘的數(shù)據(jù)需要對(duì)應(yīng)，因此，還需要標(biāo)記其位置，因此有如下設(shè)計(jì)： 矩陣A: = {key:“目標(biāo)矩陣的行序, 目標(biāo)矩陣的列序”, value:“源矩陣a, 目標(biāo)矩陣的行序, 源矩陣a的值”} 如：{key:”1,1”, value:”a,1,1”} 矩陣B: = { key:“目標(biāo)矩陣的行序, 目標(biāo)矩陣的列序”, value:“源矩陣b, 標(biāo)矩陣的列序, 源矩陣b的值”} 如：{key:”1,1”, value:”b

5、,1,10”}。 因此，該格式也是Mapper傳遞給Shuffle的鍵值對(duì)格式, Shuffle再將相同key的元素合并，得到計(jì)算Ci,j的所有數(shù)據(jù)及計(jì)算順序，Mapper讀取鍵值對(duì)，從迭代器中讀取出每一項(xiàng)value，從而計(jì)算Ci,j。 Mapper讀矩陣所在的文件，得到的鍵值對(duì)格式為： 矩陣A: = {key：“行號(hào)”，value：“i, j, Aij”} 矩陣B: = {key：“行號(hào)”，value：“i, j, Bij”} 如果要得到Mapper的輸出格式： = {key:“目標(biāo)矩陣的行序,

6、目標(biāo)矩陣的列序”, value:“源矩陣a, 目標(biāo)矩陣的行序, 源矩陣a的值”} 只需要得到目標(biāo)矩陣的列序即可輸出，而列序可以通過(guò)循環(huán)得到，因?yàn)樵撛匦枰獏⑴c目標(biāo)矩陣對(duì)應(yīng)行的所有目標(biāo)列的計(jì)算。 3.2 算法圖示 矩陣在文件中的存儲(chǔ)格式如下： 第一行為矩陣的形狀 在mapper之前，先判斷兩個(gè)矩陣是否能夠進(jìn)行乘法運(yùn)算。 3.3 代碼 一、矩陣乘法代碼如下： package main.matrix; import java.io.BufferedReader; import java.io.File; import java.io.FileInputStr

7、eam; import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStreamReader; import java.util.HashMap; import java.util.Iterator; import java.util.Map; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hado

8、op.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Ma

9、pper; 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;

10、import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class Matrix_Multi extends Configured implements Tool{ public static class MatrixMapper extends Mapper { p

11、rivate String flag = null; static int m = 0;// 矩陣A的行數(shù) private int n = 0;// 矩陣A的列數(shù) static int p = 0;// 矩陣B的列數(shù) private int q = 0;// 矩陣B的行數(shù) private int a=0, b=0;//矩陣A，B文件的行數(shù)記錄器 //計(jì)數(shù)器 enum Counter { LINESKIP, //出錯(cuò)的行 } @Override protected void setup(Context co

12、ntext) throws IOException,InterruptedException { FileSplit split = (FileSplit) context.getInputSplit(); flag = split.getPath().getName(); } /** * map階段：讀取矩陣a和b， 將矩陣中的元素按指定格式存入上下文 * 得到的鍵值對(duì)格式為： key: value_1, value_2 ... value_n */ @Override protected void map(LongWr

13、itable key, Text value, Context context)throws IOException, InterruptedException { try{ String[] val = value.toString().split(","); System.out.println("map:\t" + flag + "\t" + value.toString()); if ("test_ma.txt".equals(flag)) { a ++ ; if(a == 1){ //第一行為矩陣形狀

14、 return; } //矩陣B的列數(shù)循環(huán)，表示當(dāng)前元素要參與運(yùn)算的次數(shù)以及運(yùn)算的目標(biāo)位置 //寫(xiě)入上下文的鍵值對(duì)格式為{ key:“目標(biāo)矩陣的行序, 目標(biāo)矩陣的列序”, value:"源矩陣a, 目標(biāo)矩陣的行序, 源矩陣a的值"} for (int i = 1; i <= p; i++) { context.write(new Text(val[0] + "," + i), new Text("a,"+ val[1] + "," + val[2])); //寫(xiě)入到上下文中的形式{key:"行號(hào)"} Sys

15、tem.out.println("key: " + new Text(val[0] + "," + i).toString() + "\tvalue: " + new Text("a,"+ val[1] + "," + val[2]).toString()); } } else if ("test_mb.txt".equals(flag)) { b ++ ; if(b == 1){ //第一行為矩陣形狀 return; } //矩陣A的行數(shù)循環(huán)，表示當(dāng)前元素要參與運(yùn)算的次數(shù)以及運(yùn)算的目標(biāo)位置

16、 //寫(xiě)入上下文的鍵值對(duì)格式為{ key:“目標(biāo)矩陣的行序, 目標(biāo)矩陣的列序”, value:"源矩陣b, 標(biāo)矩陣的列序, 源矩陣b的值"} for (int i = 1; i <= m; i++) { context.write(new Text(i + "," + val[1]), new Text("b,"+ val[0] + "," + val[2])); System.out.println("key: " + new Text(i + "," + val[1]).toString() + "\tvalue: " + new Tex

17、t("b,"+ val[0] + "," + val[2]).toString()); } } }catch(java.lang.ArrayIndexOutOfBoundsException e){ // 記錄不符合輸入規(guī)則的行 context.getCounter(Counter.LINESKIP).increment(1); return; } } } public static class MatrixReducer extends Reducer

18、ble> { /** * reduce階段：從上下文讀取鍵值對(duì)，key: value_1, value_2 ... value_n * key為目標(biāo)矩陣的元素位置 * 將key對(duì)應(yīng)的values進(jìn)行整理，將矩陣A的元素放入mapA，矩陣B的元素放入mapB * 矩陣mapA的鍵值對(duì)格式為{key:“目標(biāo)矩陣的行序”, value:“源矩陣A的值”} * 矩陣mapB的鍵值對(duì)格式為{key:“目標(biāo)矩陣的列序”, value:"源矩陣B的值"} * mapA和mapB中key相同的value相乘，在相加，得到目標(biāo)矩陣在該位置的值

19、 */ @Override protected void reduce(Text key, Iterable values, Context context)throws IOException, InterruptedException { Map mapA = new HashMap(); Map mapB = new HashMap(); for (Text value : values) {

20、System.out.println("reduce:\t" +value.toString()); String[] val = value.toString().split(","); if ("a".equals(val[0])) { mapA.put(val[1], val[2]); } else if ("b".equals(val[0])) { mapB.put(val[1], val[2]); } } int result = 0; Iterator mKeys

21、 = mapA.keySet().iterator(); while (mKeys.hasNext()) { String mkey = mKeys.next(); if (mapB.get(mkey) == null) {// 因?yàn)閙key取的是mapA的key集合，所以只需要判斷mapB是否存在即可。 continue; } result += Integer.parseInt(mapA.get(mkey)) * Integer.parseInt(mapB.get(mkey))

22、; } context.write(key, new IntWritable(result)); } } public boolean init(Configuration conf, String input1, String input2) throws IOException { FSDataInputStream in1 = null; FSDataInputStream in2 = null; BufferedReader d1 = null; BufferedReader d2 = null; FileSyste

23、m fs = FileSystem.get(conf); Path ma = new Path(input1); Path mb = new Path(input2); try{ in1 = fs.open(ma); d1 = new BufferedReader(new InputStreamReader(in1)); String line_a = d1.readLine(); System.out.println("矩陣A的形狀：" + line_a); in2 = fs.open(mb); d2 = new B

24、ufferedReader(new InputStreamReader(in2)); String line_b = d2.readLine(); System.out.println("矩陣B的形狀：" + line_b); int c_a = Integer.parseInt(line_a.split(",")[1]); int r_b = Integer.parseInt(line_b.split(",")[0]); if(c_a != r_b){ System.out.println(c_a + "," + r_b + " "

25、 + "無(wú)法相乘"); return false; } // 設(shè)置目標(biāo)矩陣形狀 MatrixMapper.m = Integer.parseInt(line_a.split(",")[0]); MatrixMapper.p = Integer.parseInt(line_b.split(",")[1]); }catch(Exception e){ e.printStackTrace(); return false; }finally{ d1.close(); d2.close(); in1.close

26、(); in2.close(); fs.close(); } return true; } /** * 設(shè)置作業(yè) */ @Override public int run(String[] args)throws Exception { Configuration conf = getConf(); String input1 = args[0]; String input2 = args[1]; String output = args[2]; if(!init(conf, input1, input

27、2)) { return 0; } /* String input1 = "hdfs://nna:9000/in/ma.txt"; String input2 = "hdfs://nna:9000/in/mb.txt"; String output = "hdfs://nna:9000/out"; */ conf.addResource("/usr/hadoop/etc/hadoop/core-site.xml"); conf.addResource("/usr/hadoop/etc/hadoop/hdf

28、s-site.xml"); conf.addResource("/usr/hadoop/etc/hadoop/mapred-site.xml"); conf.addResource("/usr/hadoop/etc/hadoop/yarn-site.xml"); Job job = Job.getInstance(conf, "MatrixMul"); //實(shí)例化一個(gè)作業(yè) job.setJarByClass(Matrix_Multi.class); //指定主類(lèi) job.setOutputKeyClass(Text.cla

29、ss); //指定輸出的key格式 job.setOutputValueClass(Text.class); //指定輸出的value格式 job.setMapperClass(MatrixMapper.class); //指定Mapper類(lèi) job.setReducerClass(MatrixReducer.class); //指定Reducer類(lèi) job.setInputFormatClass(TextInputFormat.class); //指定輸入的數(shù)據(jù)類(lèi)型 job.se

30、tOutputFormatClass(TextOutputFormat.class); //指定輸出的數(shù)據(jù)類(lèi)型 FileInputFormat.setInputPaths(job, new Path(input1), new Path(input2));//指定輸入路徑 Path outputPath = new Path(output); //指定輸出路徑 outputPath.getFileSystem(conf).delete(outputPath, true); //若輸出路徑存在則刪除 FileOutputFo

31、rmat.setOutputPath(job, outputPath); //將該作業(yè)的結(jié)果寫(xiě)入輸出路徑 return job.waitForCompletion(true) ? 0 : 1; //等待任務(wù)完成，完成返回0，未完成返回1 } /** * @param args * @throws Exception */ public static void main(String[] args)throws Exception { // TODO Auto-generated method stub Con

32、figuration conf = new Configuration(); conf.set("fs.default.name", "hdfs://nna:9000"); //設(shè)置文件系統(tǒng) long startTime = System.currentTimeMillis(); int res = ToolRunner.run(conf,new Matrix_Multi(),args); long endTime = System.currentTimeMillis(); if(res == 0){ System.out.println("Succ

33、essful!!!"); System.out.println("執(zhí)行時(shí)間(ms)：" + (endTime - startTime)); } else{ System.out.println("Error!!!"); } System.exit(res); } } 二、自動(dòng)生成隨機(jī)數(shù)矩陣代碼如下： package main.matrix; import java.io.BufferedWriter; import java.io.File; import java.io.FileWriter; import java.i

34、o.IOException; import java.util.Random; import java.util.Scanner; public class Create_matrix { private int m; //矩陣行數(shù) private int n; //矩陣列數(shù) Create_matrix(int m, int n){ this.m = m; this.n = n; } public int create(String path) throws IOException{ BufferedWriter write

35、r = null; try{ //打開(kāi)文件 File file = new File(path); if(!file.exists()){ file.createNewFile(); } writer = new BufferedWriter(new FileWriter(file)); String line = "" + m + "," + n + "\n"; //首行寫(xiě)入矩陣形狀 //將該字符串寫(xiě)入文件 writer.write(line); writer.flush();

36、//創(chuàng)建隨機(jī)矩陣 Random random = new Random(); int i, j, number; for(i=1; i<=m; i++){ for(j=1; j<=n; j++){ number = random.nextInt(100); line = "" + i + "," + j + "," + number + "\n"; //將該字符串寫(xiě)入文件 writer.write(line); writer.flush(); } } }catch(Exce

37、ption e){ e.printStackTrace(); }finally{ writer.close(); } System.out.println("complete!"); return 0; } public static void main(String args[]) throws IOException{ int m,n; Scanner sc = new Scanner(System.in); String path_ma = "/home/gaid/file/test_ma.txt"; S

38、tring path_mb = "/home/gaid/file/test_mb.txt"; System.out.println("請(qǐng)輸入矩陣A的行列(m n):"); String m_n = sc.nextLine(); m = Integer.parseInt(m_n.split(" ")[0]); n = Integer.parseInt(m_n.split(" ")[1]); Create_matrix ma = new Create_matrix(m, n); ma.create(path_ma); m_n = sc.

39、nextLine(); m = Integer.parseInt(m_n.split(" ")[0]); n = Integer.parseInt(m_n.split(" ")[1]); Create_matrix mb = new Create_matrix(m, n); mb.create(path_mb); sc.close(); } } 3.4 測(cè)試 一、10^3矩陣乘法測(cè)試 生成矩陣： 將項(xiàng)目導(dǎo)成JAR包，并提交到hadoop運(yùn)行： 執(zhí)行過(guò)程如下： Map worker和Reduce worker的實(shí)時(shí)數(shù)量如下： 運(yùn)行時(shí)間如下： 運(yùn)行結(jié)果如下： Web端查看執(zhí)行過(guò)程如下： 二、10^4矩陣乘法測(cè)試 HDFS中的文件存儲(chǔ)如下： 執(zhí)行過(guò)程如下： Map worker和Reduce worker的實(shí)時(shí)數(shù)量如下： 消耗的時(shí)間如下： 三、10^5矩陣乘法測(cè)試 HDFS存儲(chǔ)空間無(wú)法滿(mǎn)足10^5次方矩陣的乘法所需要的存儲(chǔ)空間，未做測(cè)試。 專(zhuān)心---專(zhuān)注---專(zhuān)業(yè)