矩陣乘法的mapreduce程序實現

map函數：對於矩陣M中的每一個元素m(ij)，產生一系列的key-value對<(i，k)，（M，j，m(ij)）>

其中k=1,2.....知道矩陣N的總列數;對於矩陣N中的每一個元素n(jk)，產生一系列的key-value對<(i , k) , (N , j ,n(jk)>, 其中i=1,2.......直到i=1,2.......直到矩陣M的總列數。

map

package com.cb.matrix;

import static org.mockito.Matchers.intThat;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.Mapper;

import com.sun.org.apache.bcel.internal.generic.NEW;


public class MatrixMapper extends Mapper<Object, Text, Text, Text> {
	private Text map_key=new Text();
	private Text map_value= new Text();
	private int columnN;
	private int rowM;
	/**
	 * 執行map()函數前先由conf.get()獲得main函數中提供的必要變量
	 * 也就是從輸入文件名中獲得的矩陣維度信息
	 */
	
	@Override
	protected void setup(Mapper<Object, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		Configuration config=context.getConfiguration();
		columnN=Integer.parseInt(config.get("columnN"));
		rowM =Integer.parseInt(config.get("rowM"));
	}
	
	@Override
	protected void map(Object key, Text value, Mapper<Object, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		//獲得文件名，從而區分輸入矩陣M和N
		FileSplit fileSplit=(FileSplit)context.getInputSplit();
		String fileName=fileSplit.getPath().getName();
		
		if (fileName.contains("M")) {
			String[] tuple =value.toString().split(",");
			int i =Integer.parseInt(tuple[0]);
			String[] tuples=tuple[1].split("\t");
			int j=Integer.parseInt(tuples[0]);
			int Mij=Integer.parseInt(tuples[1]);
			for(int k=1;k<columnN+1;k++){
				map_key.set(i+","+k);
				map_value.set("M"+","+j+","+Mij);
				context.write(map_key, map_value);
			}
			
		}
		else if(fileName.contains("N")){
			String[] tuple=value.toString().split(",");
			int j=Integer.parseInt(tuple[0]);
			String[] tuples =tuple[1].split("\t");
			int k=Integer.parseInt(tuples[0]);
			int Njk=Integer.parseInt(tuples[1]);
			for(int i=1;i<rowM+1;i++){
				map_key.set(i+","+k);
				map_value.set("N"+","+j+","+Njk);
				context.write(map_key, map_value);
			}
		}
		
	}

}

reduce函數：對於每一個鍵（i,k）相關聯的值（M,j,m(ij)）及（N,j,n(jk)）,根據相同的j值將m(ij)和n(jk)分別存入不一樣的數組中，而後將倆者的第j個元素抽取出來分別相乘，最後相加，便可獲得p(jk)的值。

reducer

package com.cb.matrix;


import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;



public class MatrixReducer extends Reducer<Text, Text, Text, Text> {
	private int sum=0;
	private int columnM;
	@Override
	protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		Configuration conf =context.getConfiguration();
		columnM=Integer.parseInt(conf.get("columnM"));
	}
	@Override
	protected void reduce(Text arg0, Iterable<Text> arg1, Reducer<Text, Text, Text, Text>.Context arg2)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		int[] M=new int[columnM+1];
		int[] N=new int[columnM+1];
		
		for(Text val:arg1){
			String[] tuple=val.toString().split(",");
			if(tuple[0].equals("M")){
				M[Integer.parseInt(tuple[1])]=Integer.parseInt(tuple[2]);
				
			}else{
				N[Integer.parseInt(tuple[1])]=Integer.parseInt(tuple[2]);
			}
			for(int j=1;j<columnM+1;j++){
				sum+=M[j]*N[j];
			}
			arg2.write(arg0, new Text(Integer.toString(sum)));
			sum=0;
		}
	}

}